Merge main into gatekeeper/fix-2325-1

Retain compaction mode exports alongside the upstream RTE statistics rename.

Validation: rebuilt the Python extension and passed all 25 optimize tests; Rust formatting and Python source lint/format checks passed. Full Ruff still reports 49 inherited documentation-example errors.
This commit is contained in:
Gatefixer
2026-10-20 17:22:15 +00:00
214 changed files with 7318 additions and 2184 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.41.0-beta.7"
version = "0.41.0-beta.10"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
+2
View File
@@ -8,6 +8,8 @@ A Python library for [LanceDB](https://github.com/lancedb/lancedb).
pip install lancedb
```
If no wheel is published for your platform, pip builds `lancedb` from the source distribution. That build needs a [Rust toolchain](https://rustup.rs/) and [protoc](https://grpc.io/docs/protoc-installation/) (the Protocol Buffers compiler) on your `PATH`, and can take a while.
### Pre-Haswell x86_64 hosts: `lancedb-compat`
The default `lancedb` wheel targets `x86-64-haswell` (AVX2 + FMA + F16C) for full performance on modern hardware. Pre-Haswell hosts — Intel Sandy Bridge / Ivy Bridge / Westmere; AMD Bulldozer / Piledriver / Steamroller — don't have AVX2 and crash with `Illegal instruction` at `import lancedb`.
+2 -2
View File
@@ -15,7 +15,7 @@ from ._lancedb import FtsToken
from ._lancedb import FunctionErrorFragment as FunctionErrorFragment
from ._lancedb import FunctionErrorRecord as FunctionErrorRecord
from ._lancedb import FunctionErrors as FunctionErrors
from ._lancedb import MemWalWriteSpec
from ._lancedb import RTEWriteSpec
from ._lancedb import tokenize as _tokenize
from .common import URI, sanitize_uri
from urllib.parse import urlparse
@@ -611,7 +611,7 @@ __all__ = [
"SqlQuery",
"LanceDBConnection",
"LanceNamespaceDBConnection",
"MemWalWriteSpec",
"RTEWriteSpec",
"RemoteDBConnection",
"Session",
"Table",
+24 -28
View File
@@ -539,17 +539,17 @@ class Table:
async def initial_storage_options(self) -> Optional[Dict[str, str]]: ...
async def latest_storage_options(self) -> Optional[Dict[str, str]]: ...
async def set_unenforced_primary_key(self, columns: List[str]) -> None: ...
async def enable_memwal(self, spec: MemWalWriteSpec) -> None: ...
async def disable_memwal(self) -> None: ...
async def get_memwal_spec(self) -> Optional[MemWalWriteSpec]: ...
async def enable_rte(self, spec: RTEWriteSpec) -> None: ...
async def disable_rte(self) -> None: ...
async def get_rte_spec(self) -> Optional[RTEWriteSpec]: ...
# WAL-PK-FUSION: delete both.
def hybrid_pk_fusion_learned(self) -> bool: ...
def note_hybrid_pk_fusion(self) -> None: ...
async def checkpoint_memwal(self) -> None: ...
async def flush_memwal(self) -> None: ...
async def compact_memwal(self) -> None: ...
async def get_memwal_stats(self, include_sstable_rows: bool) -> Optional[dict]: ...
async def close_memwal_writers(self) -> None: ...
async def checkpoint_rte(self) -> None: ...
async def flush_rte(self) -> None: ...
async def compact_rte(self) -> None: ...
async def get_rte_stats(self, include_sstable_rows: bool) -> Optional[dict]: ...
async def close_rte_writers(self) -> None: ...
@property
def tags(self) -> Tags: ...
@property
@@ -662,7 +662,7 @@ class Query:
def fast_search(self): ...
def with_row_id(self): ...
def postfilter(self): ...
def use_memwal(self, enable: bool): ...
def use_rte(self, enable: bool): ...
def nearest_to(self, query_vec: pa.Array) -> VectorQuery: ...
def nearest_to_text(self, query: dict) -> FTSQuery: ...
def order_by(self, ordering: Optional[List[ColumnOrdering]]): ...
@@ -679,7 +679,7 @@ class Query:
class TakeQuery:
def select(self, columns: List[str]): ...
def with_row_id(self): ...
def use_memwal(self, enable: bool): ...
def use_rte(self, enable: bool): ...
async def output_schema(self) -> pa.Schema: ...
async def execute(self) -> RecordBatchStream: ...
async def explain_plan(self, verbose: Optional[bool]) -> str: ...
@@ -698,7 +698,7 @@ class FTSQuery:
def fast_search(self): ...
def with_row_id(self): ...
def postfilter(self): ...
def use_memwal(self, enable: bool): ...
def use_rte(self, enable: bool): ...
def get_query(self) -> str: ...
def add_query_vector(self, query_vec: pa.Array) -> None: ...
def nearest_to(self, query_vec: pa.Array) -> HybridQuery: ...
@@ -726,7 +726,7 @@ class VectorQuery:
def column(self, column: str): ...
def distance_type(self, distance_type: str): ...
def postfilter(self): ...
def use_memwal(self, enable: bool): ...
def use_rte(self, enable: bool): ...
def refine_factor(self, refine_factor: int): ...
def nprobes(self, nprobes: int): ...
def search_effort(self, search_effort: Optional[float]): ...
@@ -751,7 +751,7 @@ class HybridQuery:
def fast_search(self): ...
def with_row_id(self): ...
def postfilter(self): ...
def use_memwal(self, enable: bool): ...
def use_rte(self, enable: bool): ...
def distance_type(self, distance_type: str): ...
def refine_factor(self, refine_factor: int): ...
def nprobes(self, nprobes: int): ...
@@ -779,7 +779,7 @@ class PyQueryRequest:
select_source_columns: Optional[Dict[str, str]]
fast_search: Optional[bool]
with_row_id: Optional[bool]
use_memwal: Optional[bool]
use_rte: Optional[bool]
column: Optional[str]
query_vector: Optional[List[pa.Array]]
nprobes: Optional[int]
@@ -840,26 +840,22 @@ class FtsToken:
text: str
position: int
class MemWalWriteSpec:
"""Specification selecting Lance's MemWAL LSM-style write path for
`merge_insert`."""
class RTEWriteSpec:
"""Specification enabling the realtime engine (RTE), which routes
`merge_insert` through Lance's MemWAL LSM-style write path."""
@staticmethod
def bucket(column: str, num_buckets: int) -> "MemWalWriteSpec": ...
def bucket(column: str, num_buckets: int) -> "RTEWriteSpec": ...
@staticmethod
def identity(column: str) -> "MemWalWriteSpec": ...
def identity(column: str) -> "RTEWriteSpec": ...
@staticmethod
def unsharded() -> "MemWalWriteSpec": ...
def with_maintained_indexes(
self, indexes: Optional[List[str]]
) -> "MemWalWriteSpec":
"""Set which indexes the MemWAL keeps up to date. None resolves every
def unsharded() -> "RTEWriteSpec": ...
def with_maintained_indexes(self, indexes: Optional[List[str]]) -> "RTEWriteSpec":
"""Set which indexes the RTE keeps up to date. None resolves every
index on the table at install, failing if one cannot be maintained;
a list is verbatim, empty means none."""
...
def with_writer_config_defaults(
self, defaults: Dict[str, str]
) -> "MemWalWriteSpec":
def with_writer_config_defaults(self, defaults: Dict[str, str]) -> "RTEWriteSpec":
"""Return a copy of this spec recording the given default
`ShardWriter` configuration in the MemWAL index."""
...
@@ -873,7 +869,7 @@ class MemWalWriteSpec:
def num_buckets(self) -> Optional[int]: ...
@property
def maintained_indexes(self) -> Optional[List[str]]:
"""Indexes the MemWAL keeps up to date, or None for every supported one."""
"""Indexes the RTE keeps up to date, or None for every supported one."""
...
@property
def writer_config_defaults(self) -> Dict[str, str]: ...
+19 -19
View File
@@ -1,14 +1,14 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""WAL-PK-FUSION: hybrid search's primary-key fallback for MemWAL tables.
"""WAL-PK-FUSION: hybrid search's primary-key fallback for RTE tables.
Hybrid search joins its vector and full-text legs on `_rowid`, which a MemWAL
Hybrid search joins its vector and full-text legs on `_rowid`, which an RTE
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
This is a stopgap until RTE 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`.
@@ -69,21 +69,21 @@ def _pk_columns(schema: pa.Schema) -> List[str]:
# 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,
# the fallback stops firing and hybrid on an RTE 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 "
"hybrid search on an RTE 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_memwal(False) to read the base table only (results "
"will exclude un-compacted MemWAL data)"
"the primary key. Set use_rte(False) to read the base table only (results "
"will exclude un-compacted RTE 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."""
RTE-backed, as opposed to any other failure the query might hit."""
return _WAL_ROW_ID_REFUSAL in str(error)
@@ -112,7 +112,7 @@ def _stamp_surrogate_row_ids(
"""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
group by in a reranker, and to restore the original scores — so on an RTE
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)
@@ -124,7 +124,7 @@ def _stamp_surrogate_row_ids(
"""
if not pk_columns:
raise ValueError(
"hybrid search on a MemWAL table needs an unenforced primary key to "
"hybrid search on an RTE table needs an unenforced primary key to "
"deduplicate on, and this table declares none"
)
@@ -151,13 +151,13 @@ def _stamp_surrogate_row_ids(
return stamped
def pk_fusion_learned(table, use_memwal: Optional[bool]) -> bool:
def pk_fusion_learned(table, use_rte: 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_memwal is False:
if table is None or use_rte is False:
return False
# Only an unambiguous True forks to the key: a table type that answers the
# question loosely keeps the `_rowid` path.
@@ -165,7 +165,7 @@ def pk_fusion_learned(table, use_memwal: Optional[bool]) -> bool:
class PkFusion:
"""How the legs of a hybrid query on a MemWAL table are joined: on a
"""How the legs of a hybrid query on an RTE table are joined: on a
surrogate `_rowid` built from the primary key."""
def __init__(self, pk_columns: List[str]):
@@ -190,16 +190,16 @@ class PkFusion:
# 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: "
"hybrid search cannot project a blob column on an RTE table: "
"fetching blobs needs a real _rowid, and the fresh tier has no "
"stable one. Set use_memwal(False) to read the base table only "
"(results will exclude un-compacted MemWAL data)"
"stable one. Set use_rte(False) to read the base table only "
"(results will exclude un-compacted RTE 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."""
the `_rowid` an RTE table would refuse."""
columns, self.injected = _with_pk_columns(columns, self.pk_columns)
return columns
@@ -226,7 +226,7 @@ class PkFusion:
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
"""Re-raise `error` unless it is an RTE `_rowid` refusal worth retrying on
the primary key."""
if fused or not _is_wal_row_id_refusal(error):
raise error
@@ -248,7 +248,7 @@ def with_pk_fallback(
`_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
find out: only an RTE 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.
+7 -1
View File
@@ -3,7 +3,12 @@
# ruff: noqa: F401
from .base import EmbeddingFunction, EmbeddingFunctionConfig, TextEmbeddingFunction
from .base import (
EmbeddingFunction,
EmbeddingFunctionConfig,
TextEmbeddingFunction,
UntrustedEmbeddingConfigError,
)
from .bedrock import BedRockText
from .cohere import CohereEmbeddingFunction
from .gemini_text import GeminiText
@@ -29,6 +34,7 @@ __all__ = [
"EmbeddingFunction",
"EmbeddingFunctionConfig",
"TextEmbeddingFunction",
"UntrustedEmbeddingConfigError",
"EmbeddingFunctionRegistry",
"get_registry",
"register",
+96 -2
View File
@@ -3,7 +3,7 @@
from abc import ABC, abstractmethod
import copy
from typing import List, Union
from typing import List, Optional, Tuple, Union
from lancedb.util import add_note
import numpy as np
@@ -13,6 +13,18 @@ from pydantic import BaseModel, Field, PrivateAttr
from .utils import TEXT, retry_with_exponential_backoff
class UntrustedEmbeddingConfigError(PermissionError):
"""
Raised when using an embedding function restored from table metadata whose
settings this process has not approved.
See
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code]
and
[allow_endpoint][lancedb.embeddings.EmbeddingFunctionRegistry.allow_endpoint].
"""
class EmbeddingFunction(BaseModel, ABC):
"""
An ABC for embedding functions.
@@ -37,11 +49,29 @@ class EmbeddingFunction(BaseModel, ABC):
"""
Create an instance of the embedding function
"""
from .registry import EmbeddingFunctionRegistry
resolved_kwargs = cls.__resolveVariables(kwargs)
instance = cls(**resolved_kwargs)
instance._original_args = kwargs
instance._original_args = dict(kwargs)
if (
instance.remote_code_model() is not None
and "trust_remote_code" in cls.model_fields
):
# Record the flag even when left at its default, so a missing flag
# in table metadata only ever means a table written before this.
instance._original_args.setdefault(
"trust_remote_code", instance.trust_remote_code
)
# The caller wrote this config, so this process trusts it. Without this,
# a table created with it could not be reopened in the same process.
EmbeddingFunctionRegistry.get_instance()._trust(instance)
return instance
@classmethod
def _resolve_variables(cls, args: dict) -> dict:
return cls.__resolveVariables(args)
@classmethod
def __resolveVariables(cls, args: dict) -> dict:
"""
@@ -92,6 +122,38 @@ class EmbeddingFunction(BaseModel, ABC):
"""
return []
def remote_code_model(self) -> Optional[Tuple[str, Optional[str]]]:
"""
Return the model whose repository code this function runs when its
`trust_remote_code` field is enabled, as a `(name, revision)` pair.
Return None if the function never runs code from a model repository.
When the function is restored from table metadata, `trust_remote_code`
is only enabled for models allowed with
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code].
"""
return None
@classmethod
def _legacy_trust_remote_code_default(cls) -> bool:
"""
The `trust_remote_code` to use for table metadata that omits it. Only
tables written before the flag was always recorded omit it, so this is
the default from that time.
"""
field = cls.model_fields.get("trust_remote_code")
return bool(field.default) if field is not None else False
def endpoints(self) -> List[str]:
"""
Return the non-default URLs this function sends data or credentials to.
When the function is restored from table metadata, each of these must be
allowed with
[allow_endpoint][lancedb.embeddings.EmbeddingFunctionRegistry.allow_endpoint].
"""
return []
@abstractmethod
def compute_query_embeddings(self, *args, **kwargs) -> list[Union[np.array, None]]:
"""
@@ -233,3 +295,35 @@ class TextEmbeddingFunction(EmbeddingFunction):
) -> list[Union[np.array, None]]:
"""Generate the embeddings for the given texts"""
pass
class UntrustedEmbeddingFunction(EmbeddingFunction):
"""
Stands in for an embedding function restored from table metadata whose
settings this process has not approved.
The table stays usable for anything that doesn't compute embeddings, while
computing embeddings raises
[UntrustedEmbeddingConfigError][lancedb.embeddings.UntrustedEmbeddingConfigError].
The real function is never constructed, since some load their model in
`__init__`.
"""
function_name: str
reasons: Tuple[str, ...]
def _error(self) -> UntrustedEmbeddingConfigError:
details = "\n".join(f" - {reason}" for reason in self.reasons)
return UntrustedEmbeddingConfigError(
f"Embedding function '{self.function_name}' was restored from table "
"metadata with settings this process has not approved:\n" + details
)
def compute_query_embeddings(self, *args, **kwargs):
raise self._error()
def compute_source_embeddings(self, *args, **kwargs):
raise self._error()
def ndims(self) -> int:
raise self._error()
+18 -1
View File
@@ -14,6 +14,9 @@ if TYPE_CHECKING:
import ollama
DEFAULT_OLLAMA_HOST = "http://localhost:11434"
@register("ollama")
class OllamaEmbeddings(TextEmbeddingFunction):
"""
@@ -24,11 +27,25 @@ class OllamaEmbeddings(TextEmbeddingFunction):
"""
name: str = "nomic-embed-text"
host: str = "http://localhost:11434"
host: str = DEFAULT_OLLAMA_HOST
options: Optional[dict] = None # type = ollama.Options
keep_alive: Optional[Union[float, str]] = None
ollama_client_kwargs: Optional[dict] = {}
def endpoints(self) -> List[str]:
urls = []
if self.host.rstrip("/") != DEFAULT_OLLAMA_HOST:
urls.append(self.host)
# These kwargs go to the underlying httpx client, where a proxy would
# see every request.
for key in ("proxy", "proxies"):
proxy = (self.ollama_client_kwargs or {}).get(key)
if isinstance(proxy, str):
urls.append(proxy)
elif isinstance(proxy, dict):
urls.extend(v for v in proxy.values() if isinstance(v, str))
return urls
def ndims(self) -> int:
return len(self.generate_embeddings(["foo"])[0])
@@ -58,6 +58,9 @@ class OpenAIEmbeddings(TextEmbeddingFunction):
def sensitive_keys():
return ["api_key"]
def endpoints(self) -> List[str]:
return [self.base_url] if self.base_url else []
@staticmethod
def model_names():
return [
+131 -6
View File
@@ -2,9 +2,15 @@
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
import json
from typing import Dict, Optional, Type
from typing import Dict, Optional, Set, Tuple, Type
from .base import EmbeddingFunction, EmbeddingFunctionConfig
from pydantic import TypeAdapter
from .base import (
EmbeddingFunction,
EmbeddingFunctionConfig,
UntrustedEmbeddingFunction,
)
class EmbeddingFunctionRegistry:
@@ -42,6 +48,9 @@ class EmbeddingFunctionRegistry:
def __init__(self):
self._functions = {}
self._variables = {}
# Revision is a str, None for the default branch, or _ANY_REVISION.
self._allowed_remote_code: Set[Tuple[str, object]] = set()
self._allowed_endpoints: Set[str] = set()
def register(self, alias: Optional[str] = None):
"""
@@ -122,20 +131,129 @@ class EmbeddingFunctionRegistry:
obj["vector_column"]: EmbeddingFunctionConfig(
vector_column=obj["vector_column"],
source_column=obj["source_column"],
function=self.get(obj["name"]).create(**obj["model"]),
function=self._restore(obj["name"], obj["model"]),
)
for obj in raw_list
}
def _restore(self, name: str, args: dict) -> EmbeddingFunction:
"""
Create an embedding function from table metadata. Anyone who can write
the table controls `args`, so settings that run model code or send data
elsewhere must already be trusted by this process.
"""
cls = self.get(name)
resolved = cls._resolve_variables(args)
# Inspect the config without running __init__, which may load a model.
probe = cls.model_construct(**resolved)
reasons = []
model = probe.remote_code_model()
if model is not None:
allowed = self._remote_code_allowed(*model)
explicit = "trust_remote_code" in resolved
if explicit:
requested = TypeAdapter(bool).validate_python(
resolved["trust_remote_code"]
)
else:
requested = cls._legacy_trust_remote_code_default()
# Always set the flag, since a subclass may default it to True.
resolved["trust_remote_code"] = requested and allowed
# A legacy table that omits the flag is downgraded rather than
# blocked: most sentence-transformers models don't need remote code,
# and those tables relied on the old True default without asking.
if requested and not allowed and explicit:
model_name, revision = model
reasons.append(
f"runs code from model '{model_name}' at revision "
f"{revision!r}. If you trust it, call "
f"get_registry().allow_remote_code({model_name!r}, "
f"revision={revision!r})"
)
reasons.extend(
f"sends data to endpoint '{url}'. If you trust it, call "
f"get_registry().allow_endpoint({url!r})"
for url in probe.endpoints()
if _normalize_url(url) not in self._allowed_endpoints
)
if reasons:
func = UntrustedEmbeddingFunction(
function_name=name, reasons=tuple(reasons)
)
else:
func = cls(**resolved)
func._original_args = args
return func
def allow_remote_code(self, model: str, revision: Optional[str] = None) -> None:
"""
Trust the code in a model repository when an embedding function that
loads it is restored from table metadata.
Table metadata cannot enable `trust_remote_code` by itself, since anyone
who can write to the table could then run code in this process. Embedding
functions created in this process with `trust_remote_code=True` are
allowed automatically.
Parameters
----------
model : str
The model name, such as `"nomic-ai/nomic-embed-text-v1.5"`.
revision : Optional[str]
The revision the table's embedding function must pin. Pinning a
commit hash is recommended, since the code at a branch or tag can
change after you review it. If None, any revision is trusted,
including pull request refs that anyone can push to the repository.
"""
self._allowed_remote_code.add(
(model, _ANY_REVISION if revision is None else revision)
)
def allow_endpoint(self, url: str) -> None:
"""
Trust a custom endpoint, such as an OpenAI `base_url` or Ollama `host`,
when an embedding function that uses it is restored from table metadata.
Table metadata cannot pick an endpoint by itself, since it could send
your data and API keys to a server you don't control. Embedding functions
created in this process are allowed to use their endpoints automatically.
Parameters
----------
url : str
The endpoint URL. Trailing slashes are ignored.
"""
self._allowed_endpoints.add(_normalize_url(url))
def _remote_code_allowed(self, model: str, revision: Optional[str]) -> bool:
return (model, _ANY_REVISION) in self._allowed_remote_code or (
model,
revision,
) in self._allowed_remote_code
def _trust(self, func: EmbeddingFunction) -> None:
model = func.remote_code_model()
if model is not None and getattr(func, "trust_remote_code", False):
# Only the revision this code selected, where None means the default
# branch. Other revisions, such as PR refs, may hold other code.
self._allowed_remote_code.add(model)
for url in func.endpoints():
self.allow_endpoint(url)
def function_to_metadata(self, conf: "EmbeddingFunctionConfig"):
"""
Convert the given embedding function and source / vector column configs
into a config dictionary that can be serialized into arrow metadata
"""
func = conf.function
name = getattr(
func, "__embedding_function_registry_alias__", func.__class__.__name__
)
if isinstance(func, UntrustedEmbeddingFunction):
name = func.function_name
else:
name = getattr(
func, "__embedding_function_registry_alias__", func.__class__.__name__
)
json_data = func.safe_model_dump()
return {
"name": name,
@@ -180,6 +298,13 @@ class EmbeddingFunctionRegistry:
return self._variables[name]
_ANY_REVISION = object()
def _normalize_url(url: str) -> str:
return url.rstrip("/")
# Global instance
__REGISTRY__ = EmbeddingFunctionRegistry()
@@ -2,7 +2,7 @@
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
from typing import List, Union
from typing import List, Optional, Tuple, Union
import numpy as np
@@ -27,19 +27,32 @@ class SentenceTransformerEmbeddings(TextEmbeddingFunction):
The device to use for the model
normalize : bool, default True
Whether to normalize the embeddings
trust_remote_code : bool, default True
Whether to trust the remote code
trust_remote_code : bool, default False
Whether to run code from the model's repository. Only enable this for
repositories you trust. When restored from table metadata, this is only
enabled for models allowed with
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code].
revision : Optional[str], default None
The model revision to load, such as a commit hash.
"""
name: str = "all-MiniLM-L6-v2"
device: str = "cpu"
normalize: bool = True
trust_remote_code: bool = True
trust_remote_code: bool = False
revision: Optional[str] = None
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._ndims = None
def remote_code_model(self) -> Optional[Tuple[str, Optional[str]]]:
return (self.name, self.revision)
@classmethod
def _legacy_trust_remote_code_default(cls) -> bool:
return True
@property
def embedding_model(self):
"""
@@ -84,5 +97,8 @@ class SentenceTransformerEmbeddings(TextEmbeddingFunction):
"sentence_transformers", "sentence-transformers"
)
return sentence_transformers.SentenceTransformer(
self.name, device=self.device, trust_remote_code=self.trust_remote_code
self.name,
device=self.device,
trust_remote_code=self.trust_remote_code,
revision=self.revision,
)
@@ -3,7 +3,7 @@
from functools import cached_property
from typing import List, Any
from typing import List, Any, Optional, Tuple
import numpy as np
@@ -34,7 +34,11 @@ class TransformersEmbeddingFunction(EmbeddingFunction):
Whether or not to allow for custom models defined on the HuggingFace
Hub in their own modeling files. This option should only be set to True
for repositories you trust and in which you have read the code, as it
will execute code present on the Hub on your local machine.
will execute code present on the Hub on your local machine. When
restored from table metadata, this is only enabled for models allowed with
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code].
revision : Optional[str]
The model revision to load, such as a commit hash. Default is None.
to download package, run :
`pip install transformers`
@@ -45,6 +49,7 @@ class TransformersEmbeddingFunction(EmbeddingFunction):
name: str = "colbert-ir/colbertv2.0"
device: str = "cpu"
trust_remote_code: bool = False
revision: Optional[str] = None
_tokenizer: Any = PrivateAttr()
_model: Any = PrivateAttr()
@@ -52,12 +57,17 @@ class TransformersEmbeddingFunction(EmbeddingFunction):
super().__init__(*args, **kwargs)
self._ndims = None
transformers = attempt_import_or_raise("transformers")
self._tokenizer = transformers.AutoTokenizer.from_pretrained(self.name)
self._tokenizer = transformers.AutoTokenizer.from_pretrained(
self.name, revision=self.revision
)
self._model = transformers.AutoModel.from_pretrained(
self.name, trust_remote_code=self.trust_remote_code
self.name, trust_remote_code=self.trust_remote_code, revision=self.revision
)
self._model.to(self.device)
def remote_code_model(self) -> Optional[Tuple[str, Optional[str]]]:
return (self.name, self.revision)
model_config = ConfigDict(ignored_types=(cached_property,))
def ndims(self):
@@ -304,6 +304,11 @@ def _is_non_retryable_error(error: Exception) -> bool:
Returns:
True if the error should not be retried, False otherwise
"""
from .base import UntrustedEmbeddingConfigError
if isinstance(error, UntrustedEmbeddingConfigError):
return True
# Check for OpenAI authentication errors
error_type = type(error).__name__
if error_type == "AuthenticationError":
@@ -87,6 +87,9 @@ class WatsonxEmbeddings(TextEmbeddingFunction):
url: Optional[str] = None
params: Optional[Dict] = None
def endpoints(self) -> List[str]:
return [self.url] if self.url else []
@staticmethod
def sensitive_keys():
return ["api_key"]
+2 -5
View File
@@ -697,6 +697,7 @@ _GRAMMAR_PRIMITIVES = (
(pa.string(), "utf8"),
(pa.large_string(), "large_utf8"),
(pa.binary(), "binary"),
(pa.large_binary(), "large_binary"),
(pa.date32(), "date32"),
(pa.date64(), "date64"),
)
@@ -846,11 +847,7 @@ def _exact_blob_storage_type(field: pa.Field) -> dict[str, Any]:
{
"name": child.name,
"nullable": child.nullable,
"type": (
{"type": "large_binary"}
if pa.types.is_large_binary(child.type)
else _exact_arrow_type(child.type)
),
"type": _exact_arrow_type(child.type),
}
for child in storage
],
+8 -8
View File
@@ -37,7 +37,7 @@ class LanceMergeInsertBuilder(object):
self._when_not_matched_by_source_condition_expr = None
self._timeout = None
self._use_index = True
self._use_memwal = None
self._use_rte = None
self._validate_single_shard = None
def when_matched_update_all(
@@ -115,22 +115,22 @@ class LanceMergeInsertBuilder(object):
self._use_index = use_index
return self
def use_memwal(self, enable: bool) -> LanceMergeInsertBuilder:
def use_rte(self, enable: bool) -> LanceMergeInsertBuilder:
"""
Control MemWAL routing for this merge.
Control RTE routing for this merge.
By default (unset), a `merge_insert` on a table with a MemWAL write spec is
By default (unset), a `merge_insert` on a table with an RTE write spec is
routed through Lance's MemWAL shard writer, and a table without one uses
the standard path.
Parameters
----------
enable: bool
``True`` forces MemWAL routing and errors if the table has no MemWAL
``True`` forces RTE routing and errors if the table has no RTE
write spec. ``False`` forces the standard write path even when a spec
is set.
"""
self._use_memwal = enable
self._use_rte = enable
return self
def validate_single_shard(
@@ -139,13 +139,13 @@ class LanceMergeInsertBuilder(object):
"""
Controls how an LSM merge checks that its input targets a single shard.
When a table has a MemWAL write spec, every row in a `merge_insert` call
When a table has an RTE write spec, every row in a `merge_insert` call
must route to the same shard. When `True` (the default), every row is
inspected to verify this. When `False`, only the first row is inspected
and the shard it routes to is used for the whole input — a faster path
for callers that have already pre-sharded their input.
Has no effect on tables without a MemWAL write spec.
Has no effect on tables without an RTE write spec.
Parameters
----------
+1 -1
View File
@@ -48,7 +48,7 @@ class PermutationBuilder:
By default, the permutation builder will create a single split that contains all
rows in the same order as the base table.
Tables with a MemWAL write spec are rejected: unflushed rows have no row id.
Tables with an RTE write spec are rejected: unflushed rows have no row id.
"""
self._async = async_permutation_builder(table)
+39 -39
View File
@@ -887,10 +887,10 @@ class Query(pydantic.BaseModel):
# if true, will only search the indexed data
fast_search: Optional[bool] = None
# MemWAL LSM read routing: None auto-routes when the table carries a write
# RTE LSM read routing: None auto-routes when the table carries a write
# spec, True forces the LSM scanner (errors without a spec), False reads the
# base table only
use_memwal: Optional[bool] = None
use_rte: Optional[bool] = None
# size of the nearest neighbor list maintained during HNSW search
ef: Optional[int] = None
@@ -910,9 +910,9 @@ class Query(pydantic.BaseModel):
query.full_text_query = req.full_text_search
query.columns = req.select
query.with_row_id = req.with_row_id
# use_memwal is a genuine tri-state (None / True / False); preserve it as-is
# use_rte is a genuine tri-state (None / True / False); preserve it as-is
# so a round-tripped query keeps an explicit False.
query.use_memwal = req.use_memwal
query.use_rte = req.use_rte
query.vector_column = req.column
query.vector = req.query_vector
query.distance_type = req.distance_type
@@ -1085,7 +1085,7 @@ class LanceQueryBuilder(ABC):
self._with_row_address = None
self._fragments = None
self._fragment_ids = None
self._use_memwal = None
self._use_rte = None
self._vector = None
self._text = None
self._ef = None
@@ -1520,10 +1520,10 @@ class LanceQueryBuilder(ABC):
self._fragment_ids = fragment_ids
return self
def use_memwal(self, enable: bool) -> Self:
"""Control MemWAL LSM read routing for this query.
def use_rte(self, enable: bool) -> Self:
"""Control RTE LSM read routing for this query.
By default (unset), a query against a table with a MemWAL write spec is
By default (unset), a query against a table with an RTE write spec is
routed through the LSM scanner so it also returns data written via the
``merge_insert`` LSM path that has not yet been compacted into the base
table (active/frozen memtables + SSTables); a table without a
@@ -1532,8 +1532,8 @@ class LanceQueryBuilder(ABC):
Parameters
----------
enable : bool
``True`` forces the LSM scanner and errors if the table has no MemWAL
write spec. ``False`` bypasses the MemWAL and reads the base table
``True`` forces the LSM scanner and errors if the table has no RTE
write spec. ``False`` bypasses the RTE and reads the base table
only, even when a spec is present.
Returns
@@ -1541,7 +1541,7 @@ class LanceQueryBuilder(ABC):
LanceQueryBuilder
The LanceQueryBuilder object.
"""
self._use_memwal = enable
self._use_rte = enable
return self
def explain_plan(self, verbose: Optional[bool] = False) -> str:
@@ -1878,7 +1878,7 @@ class LanceVectorQueryBuilder(LanceQueryBuilder):
probing may continue when too few candidates are found. Any explicit
effort, including 0.5, overrides nprobes in either call order. Explicit
minimum and maximum bounds still apply. This is not a recall guarantee
and does not change refinement or HNSW ef. MemWAL LSM queries do not
and does not change refinement or HNSW ef. RTE LSM queries do not
yet support this option.
Examples
@@ -2055,7 +2055,7 @@ class LanceVectorQueryBuilder(LanceQueryBuilder):
with_row_address=self._with_row_address,
fragments=self._fragments,
fragment_ids=self._fragment_ids,
use_memwal=self._use_memwal,
use_rte=self._use_rte,
offset=self._offset,
fast_search=self._fast_search,
ef=self._ef,
@@ -2280,7 +2280,7 @@ class LanceFtsQueryBuilder(LanceQueryBuilder):
with_row_address=self._with_row_address,
fragments=self._fragments,
fragment_ids=self._fragment_ids,
use_memwal=self._use_memwal,
use_rte=self._use_rte,
full_text_query=FullTextSearchQuery(
query=self._query_with_phrase_semantics(), columns=self._fts_columns
),
@@ -2362,7 +2362,7 @@ class LanceEmptyQueryBuilder(LanceQueryBuilder):
with_row_address=self._with_row_address,
fragments=self._fragments,
fragment_ids=self._fragment_ids,
use_memwal=self._use_memwal,
use_rte=self._use_rte,
offset=self._offset,
order_by=self._order_by,
)
@@ -2458,7 +2458,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
"""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
"all")` also sets for its own use — on an RTE 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
@@ -2728,7 +2728,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
if reranker.score == "all":
# 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.
# reranker's need for row ids must not refuse an RTE table.
self._with_row_id = True
return self
@@ -2766,7 +2766,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
probing may continue when too few candidates are found. Any explicit
effort, including 0.5, overrides nprobes in either call order. Explicit
minimum and maximum bounds still apply. This is not a recall guarantee
and does not change refinement or HNSW ef. MemWAL LSM queries do not
and does not change refinement or HNSW ef. RTE LSM queries do not
yet support this option.
Examples
@@ -3040,7 +3040,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
# 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_memwal):
if _wal_hybrid.pk_fusion_learned(self._table, self._use_rte):
self._pk_fusion = _wal_hybrid.PkFusion.for_query(
self._table.schema,
self._columns,
@@ -3057,9 +3057,9 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
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_memwal is not None:
self._vector_query.use_memwal(self._use_memwal)
self._fts_query.use_memwal(self._use_memwal)
if self._use_rte is not None:
self._vector_query.use_rte(self._use_rte)
self._fts_query.use_rte(self._use_rte)
if self._phrase_query:
self._fts_query.phrase_query(True)
if self._distance_type:
@@ -3723,12 +3723,12 @@ class AsyncStandardQuery(AsyncQueryBase):
self._inner.fast_search()
return self
def use_memwal(self, enable: bool) -> Self:
def use_rte(self, enable: bool) -> Self:
"""
Control MemWAL LSM read routing for this query.
Control RTE LSM read routing for this query.
By default (unset), a query against a table with a MemWAL write spec (see
[AsyncTable.enable_memwal][lancedb.table.AsyncTable.enable_memwal])
By default (unset), a query against a table with an RTE write spec (see
[AsyncTable.enable_rte][lancedb.table.AsyncTable.enable_rte])
is routed through the LSM scanner so it also returns data written via the
``merge_insert`` LSM path that has not yet been compacted into the base
table (the active/frozen in-memory memtables and the SSTables),
@@ -3737,11 +3737,11 @@ class AsyncStandardQuery(AsyncQueryBase):
Parameters
----------
enable : bool
``True`` forces the LSM scanner and errors if the table has no MemWAL
write spec. ``False`` bypasses the MemWAL and reads the base table
``True`` forces the LSM scanner and errors if the table has no RTE
write spec. ``False`` bypasses the RTE and reads the base table
only, even when a spec is present.
"""
self._inner.use_memwal(enable)
self._inner.use_rte(enable)
return self
def postfilter(self) -> Self:
@@ -4064,7 +4064,7 @@ class AsyncVectorQueryBase:
minimum and maximum bounds remain effective. Zero starts at the caller
minimum and one searches all partitions within the maximum. Later
probing may continue to obtain enough candidates. This is not a recall
guarantee and does not change refinement or HNSW ef. MemWAL LSM queries
guarantee and does not change refinement or HNSW ef. RTE LSM queries
do not yet support this option.
Examples
@@ -4388,7 +4388,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
# 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_memwal):
if _wal_hybrid.pk_fusion_learned(self._table, fts_req.use_rte):
self._pk_fusion = _wal_hybrid.PkFusion.for_query(
await self._table.schema(),
_query_request_projection(fts_req),
@@ -4577,13 +4577,13 @@ class AsyncTakeQuery(AsyncQueryBase):
def __init__(self, inner: LanceTakeQuery, table: Optional["AsyncTable"] = None):
super().__init__(inner, table)
def use_memwal(self, enable: bool) -> "AsyncTakeQuery":
"""Control MemWAL LSM read routing for this take query.
def use_rte(self, enable: bool) -> "AsyncTakeQuery":
"""Control RTE LSM read routing for this take query.
``False`` bypasses the MemWAL and reads the base table only — the escape
``False`` bypasses the RTE and reads the base table only — the escape
hatch, since take-by-row-id/offset is not supported on the LSM scanner.
"""
self._inner.use_memwal(enable)
self._inner.use_rte(enable)
return self
async def _plain_scan_to_pandas(
@@ -4644,14 +4644,14 @@ class BaseQueryBuilder(object):
self._inner.with_row_id()
return self
def use_memwal(self, enable: bool) -> Self:
def use_rte(self, enable: bool) -> Self:
"""
Control MemWAL LSM read routing for this query.
Control RTE LSM read routing for this query.
``False`` bypasses the MemWAL and reads the base table only, the escape
``False`` bypasses the RTE and reads the base table only, the escape
hatch for shapes the LSM scanner cannot honor (e.g. take-by-row-id).
"""
self._inner.use_memwal(enable)
self._inner.use_rte(enable)
return self
def with_row_address(self, with_row_address: bool = True) -> Self:
+24 -24
View File
@@ -31,7 +31,7 @@ from lancedb._lancedb import (
DropColumnsResult,
FtsToken,
IndexConfig,
MemWalWriteSpec,
RTEWriteSpec,
MergeResult,
UpdateResult,
)
@@ -1039,17 +1039,17 @@ class RemoteTable(Table):
"""Set the unenforced primary key for this table to a single column."""
return LOOP.run(self._table.set_unenforced_primary_key(columns))
def enable_memwal(self, spec: "MemWalWriteSpec") -> None:
"""Install an MemWalWriteSpec."""
return LOOP.run(self._table.enable_memwal(spec))
def enable_rte(self, spec: "RTEWriteSpec") -> None:
"""Install an RTEWriteSpec."""
return LOOP.run(self._table.enable_rte(spec))
def disable_memwal(self) -> None:
"""Remove the MemWalWriteSpec."""
return LOOP.run(self._table.disable_memwal())
def disable_rte(self) -> None:
"""Remove the RTEWriteSpec."""
return LOOP.run(self._table.disable_rte())
def get_memwal_spec(self) -> Optional["MemWalWriteSpec"]:
"""Read the installed MemWalWriteSpec, or ``None``."""
return LOOP.run(self._table.get_memwal_spec())
def get_rte_spec(self) -> Optional["RTEWriteSpec"]:
"""Read the installed RTEWriteSpec, or ``None``."""
return LOOP.run(self._table.get_rte_spec())
# WAL-PK-FUSION: delete both hooks.
def _hybrid_pk_fusion_learned(self) -> bool:
@@ -1059,31 +1059,31 @@ class RemoteTable(Table):
def _note_hybrid_pk_fusion(self) -> None:
self._table._note_hybrid_pk_fusion()
def checkpoint_memwal(self) -> None:
def checkpoint_rte(self) -> None:
"""Synchronous version of
[`AsyncTable.checkpoint_memwal`][lancedb.AsyncTable.checkpoint_memwal]."""
return LOOP.run(self._table.checkpoint_memwal())
[`AsyncTable.checkpoint_rte`][lancedb.AsyncTable.checkpoint_rte]."""
return LOOP.run(self._table.checkpoint_rte())
def flush_memwal(self) -> None:
def flush_rte(self) -> None:
"""Synchronous version of
[`AsyncTable.flush_memwal`][lancedb.AsyncTable.flush_memwal]."""
return LOOP.run(self._table.flush_memwal())
[`AsyncTable.flush_rte`][lancedb.AsyncTable.flush_rte]."""
return LOOP.run(self._table.flush_rte())
def compact_memwal(self) -> None:
def compact_rte(self) -> None:
"""Synchronous version of
[`AsyncTable.compact_memwal`][lancedb.AsyncTable.compact_memwal]."""
return LOOP.run(self._table.compact_memwal())
[`AsyncTable.compact_rte`][lancedb.AsyncTable.compact_rte]."""
return LOOP.run(self._table.compact_rte())
def get_memwal_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
def get_rte_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
"""Synchronous version of
[`AsyncTable.get_memwal_stats`][lancedb.AsyncTable.get_memwal_stats]."""
[`AsyncTable.get_rte_stats`][lancedb.AsyncTable.get_rte_stats]."""
return LOOP.run(
self._table.get_memwal_stats(include_sstable_rows=include_sstable_rows)
self._table.get_rte_stats(include_sstable_rows=include_sstable_rows)
)
def close_memwal_writers(self) -> None:
def close_rte_writers(self) -> None:
"""No-op for remote tables (no local shard writers)."""
return LOOP.run(self._table.close_memwal_writers())
return LOOP.run(self._table.close_rte_writers())
def drop_index(self, index_name: str):
return LOOP.run(self._table.drop_index(index_name))
+19 -5
View File
@@ -261,10 +261,15 @@ class StreamingDataset(IterableDataset):
storage) at the cost of higher memory usage per split buffer. Defaults
to ``DEFAULT_READ_BATCH_SIZE`` (64).
io_queue_depth:
Number of I/O batches to keep in flight per split. Higher values
overlap storage latency with transform and training compute at the cost
of more memory and threads. Must be greater than zero. Defaults to
``DEFAULT_PREFETCH_BATCHES`` (4).
Number of fetched batches to keep ahead of the transform stage per
split: this bounds the combined count of in-flight fetches and
already-fetched batches still waiting for a transform slot. Higher
values overlap storage latency with transform and training compute
at the cost of more memory and threads, but I/O stops running ahead
once this many batches are buffered, applying backpressure when the
transform stage (or ``transform_queue_depth``) is the bottleneck.
Must be greater than zero. Defaults to ``DEFAULT_PREFETCH_BATCHES``
(4).
columns:
Optional list of column names to read. When set, only those columns
are fetched from storage; all others are omitted. ``None`` (the
@@ -836,8 +841,17 @@ class StreamingDataset(IterableDataset):
io_pending[i].append((abs_start, io_pool.submit(_io_call, perm_i, indices)))
def _fill_io(i: int) -> None:
# Budget covers both in-flight fetches (io_pending) and batches
# that have already landed but are still waiting for a transform
# slot (raw_batches). Without raw_batches in the count, a slow
# transform stage never applies backpressure to Stage 1: each
# completed fetch is drained into raw_batches and immediately
# replaced by a new one, so I/O keeps racing arbitrarily far
# ahead of the transform stage instead of merely overlapping
# storage latency with it.
while (
len(io_pending[i]) < io_queue_depth and fetch_head[i] < split_sizes[i]
len(io_pending[i]) + len(raw_batches[i]) < io_queue_depth
and fetch_head[i] < split_sizes[i]
):
_submit_io(i)
+112 -90
View File
@@ -224,7 +224,7 @@ if TYPE_CHECKING:
DeleteResult,
DropColumnsResult,
FtsToken,
MemWalWriteSpec,
RTEWriteSpec,
MergeResult,
UpdateResult,
)
@@ -427,10 +427,19 @@ def _blob_input_to_arrow(data: Any, schema: Optional[pa.Schema]) -> Optional[pa.
if isinstance(field.type, pa.ExtensionType)
else field.type
)
storage = pa.array(
[_blob_value_to_storage(value) for value in values[field.name]],
type=storage_type,
)
storage_values = [_blob_value_to_storage(value) for value in values[field.name]]
missing_range_fields = {"position", "size"} - set(storage_type.names)
# Arrow silently ignores dictionary keys absent from the storage type.
if missing_range_fields and any(
value is not None
and any(value.get(name) is not None for name in missing_range_fields)
for value in storage_values
):
raise ValueError(
f"Blob column '{field.name}' storage schema cannot store byte ranges. "
"Create the column with lancedb.blob() to use position and size."
)
storage = pa.array(storage_values, type=storage_type)
column = (
pa.ExtensionArray.from_storage(field.type, storage)
if isinstance(field.type, pa.ExtensionType)
@@ -665,7 +674,7 @@ def _cast_to_target_schema(
def _preserve_blob_v2_fields(
inferred_schema: pa.Schema, existing_schema: pa.Schema
) -> pa.Schema:
"""Keep existing blob fields when an overwrite infers a replacement schema."""
"""Keep inferred blob semantics without overriding explicit replacement types."""
existing_fields = {field.name: field for field in existing_schema}
fields = [
_preserve_blob_v2_field(field, existing_fields[field.name])
@@ -677,6 +686,9 @@ def _preserve_blob_v2_fields(
def _preserve_blob_v2_field(inferred: pa.Field, existing: pa.Field) -> pa.Field:
if is_blob_v2_field(inferred):
# An explicitly declared blob layout is part of the replacement schema.
return inferred
if is_blob_v2_field(existing):
return existing
@@ -714,12 +726,10 @@ def _preserve_blob_v2_field(inferred: pa.Field, existing: pa.Field) -> pa.Field:
def _coerce_blob_write_columns(
batch: pa.RecordBatch, target_schema: pa.Schema
) -> pa.RecordBatch:
"""Materialize blob storage structs before the stream leaves Python.
"""Materialize Python bytes and URIs as blob structs before Arrow casting.
merge_insert requires its source reader to already match the table's
physical schema. Unlike add and insert, it does not pass through
LanceDB's Rust blob coercion, so preserving binary input here would
reach Lance as binary and fail the schema check.
Existing blob structs retain their input layout. Rust normalizes them to
the stored schema and rejects populated ranges on two-field targets.
"""
columns = []
fields = []
@@ -1050,6 +1060,17 @@ def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
"(Utf8, LargeUtf8, Utf8View), or a Struct with a "
"'data' or 'uri' child"
)
if pa.types.is_struct(source_type) or (
isinstance(source_type, pa.ExtensionType)
and (
not isinstance(target_field.type, pa.ExtensionType)
or type(source_type) is type(target_field.type)
)
):
# Keep the source layout until Rust can fill missing range fields or
# validate every batch before dropping all-null range fields. Arrow
# casting here would reject differing layouts before that validation.
return field.with_metadata(target_field.metadata)
# LanceDB exposes stored JSON columns as lance.json (JSONB-backed LargeBinary), but
# casting the input to that storage type here merely relabels the raw JSON bytes as
# JSONB. Lance must see arrow.json so it can perform the JSONB encoding.
@@ -1909,8 +1930,8 @@ class Table(ABC):
- pyarrow.Table or pyarrow.RecordBatch
mode: str
The mode to use when writing the data. Valid values are
"append" and "overwrite". Overwrite infers a replacement schema
but preserves existing blob v2 fields present in the new data.
"append" and "overwrite". Overwrite adopts explicitly declared blob
layouts, while inferred bytes retain the existing blob v2 fields.
on_bad_vectors: str, default "error"
What to do if any of the vectors are not the same size or contains NaNs.
One of "error", "drop", "fill".
@@ -2276,8 +2297,9 @@ class Table(ABC):
One missing or deleted row ID rejects the entire batch; no partial
results are returned.
Row IDs can be obtained from a query with ``with_row_id(True)``. This
API is currently supported only by local tables.
Row IDs can be obtained from a query with ``with_row_id(True)``.
Remote tables read the requested ranges through version-pinned blob
handles using HTTP byte-range requests.
"""
@abstractmethod
@@ -4193,8 +4215,8 @@ class LanceTable(Table):
The data to insert into the table.
mode: str
The mode to use when writing the data. Valid values are
"append" and "overwrite". Overwrite infers a replacement schema
but preserves existing blob v2 fields present in the new data.
"append" and "overwrite". Overwrite adopts explicitly declared blob
layouts, while inferred bytes retain the existing blob v2 fields.
on_bad_vectors: str, default "error"
What to do if any of the vectors are not the same size or contains NaNs.
One of "error", "drop", "fill", "null".
@@ -4926,47 +4948,47 @@ class LanceTable(Table):
[`AsyncTable.set_unenforced_primary_key`][lancedb.AsyncTable.set_unenforced_primary_key]."""
return LOOP.run(self._table.set_unenforced_primary_key(columns))
def enable_memwal(self, spec: "MemWalWriteSpec") -> None:
"""Install an MemWalWriteSpec. See
[`AsyncTable.enable_memwal`][lancedb.AsyncTable.enable_memwal]."""
return LOOP.run(self._table.enable_memwal(spec))
def enable_rte(self, spec: "RTEWriteSpec") -> None:
"""Install an RTEWriteSpec. See
[`AsyncTable.enable_rte`][lancedb.AsyncTable.enable_rte]."""
return LOOP.run(self._table.enable_rte(spec))
def disable_memwal(self) -> None:
"""Remove the MemWalWriteSpec. See
[`AsyncTable.disable_memwal`][lancedb.AsyncTable.disable_memwal]."""
return LOOP.run(self._table.disable_memwal())
def disable_rte(self) -> None:
"""Remove the RTEWriteSpec. See
[`AsyncTable.disable_rte`][lancedb.AsyncTable.disable_rte]."""
return LOOP.run(self._table.disable_rte())
def get_memwal_spec(self) -> Optional["MemWalWriteSpec"]:
"""Read the installed MemWalWriteSpec, or ``None``. See
[`AsyncTable.get_memwal_spec`][lancedb.AsyncTable.get_memwal_spec]."""
return LOOP.run(self._table.get_memwal_spec())
def get_rte_spec(self) -> Optional["RTEWriteSpec"]:
"""Read the installed RTEWriteSpec, or ``None``. See
[`AsyncTable.get_rte_spec`][lancedb.AsyncTable.get_rte_spec]."""
return LOOP.run(self._table.get_rte_spec())
def checkpoint_memwal(self) -> None:
def checkpoint_rte(self) -> None:
"""Synchronous version of
[`AsyncTable.checkpoint_memwal`][lancedb.AsyncTable.checkpoint_memwal]."""
return LOOP.run(self._table.checkpoint_memwal())
[`AsyncTable.checkpoint_rte`][lancedb.AsyncTable.checkpoint_rte]."""
return LOOP.run(self._table.checkpoint_rte())
def flush_memwal(self) -> None:
def flush_rte(self) -> None:
"""Synchronous version of
[`AsyncTable.flush_memwal`][lancedb.AsyncTable.flush_memwal]."""
return LOOP.run(self._table.flush_memwal())
[`AsyncTable.flush_rte`][lancedb.AsyncTable.flush_rte]."""
return LOOP.run(self._table.flush_rte())
def compact_memwal(self) -> None:
def compact_rte(self) -> None:
"""Synchronous version of
[`AsyncTable.compact_memwal`][lancedb.AsyncTable.compact_memwal]."""
return LOOP.run(self._table.compact_memwal())
[`AsyncTable.compact_rte`][lancedb.AsyncTable.compact_rte]."""
return LOOP.run(self._table.compact_rte())
def get_memwal_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
def get_rte_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
"""Synchronous version of
[`AsyncTable.get_memwal_stats`][lancedb.AsyncTable.get_memwal_stats]."""
[`AsyncTable.get_rte_stats`][lancedb.AsyncTable.get_rte_stats]."""
return LOOP.run(
self._table.get_memwal_stats(include_sstable_rows=include_sstable_rows)
self._table.get_rte_stats(include_sstable_rows=include_sstable_rows)
)
def close_memwal_writers(self) -> None:
"""Close cached MemWAL shard writers. See
[`AsyncTable.close_memwal_writers`][lancedb.AsyncTable.close_memwal_writers]."""
return LOOP.run(self._table.close_memwal_writers())
def close_rte_writers(self) -> None:
"""Close cached RTE shard writers. See
[`AsyncTable.close_rte_writers`][lancedb.AsyncTable.close_rte_writers]."""
return LOOP.run(self._table.close_rte_writers())
def uses_v2_manifest_paths(self) -> bool:
"""
@@ -5624,24 +5646,24 @@ class AsyncTable:
columns = list(columns)
await self._inner.set_unenforced_primary_key(columns)
async def enable_memwal(self, spec: "MemWalWriteSpec") -> None:
"""Install an MemWalWriteSpec on this table.
async def enable_rte(self, spec: "RTEWriteSpec") -> None:
"""Install an RTEWriteSpec on this table.
The spec selects Lance's MemWAL LSM-style write path for future
`merge_insert` calls. ``MemWalWriteSpec`` chooses one of three sharding
`merge_insert` calls. ``RTEWriteSpec`` chooses one of three sharding
strategies:
- ``MemWalWriteSpec.bucket(column, num_buckets)`` — hash-bucket writes by
- ``RTEWriteSpec.bucket(column, num_buckets)`` — hash-bucket writes by
the single-column unenforced primary key.
- ``MemWalWriteSpec.identity(column)`` — shard by the raw value of a
- ``RTEWriteSpec.identity(column)`` — shard by the raw value of a
scalar column.
- ``MemWalWriteSpec.unsharded()`` — route every write to a single shard.
- ``RTEWriteSpec.unsharded()`` — route every write to a single shard.
All variants require the table to have an unenforced primary key set
via [`set_unenforced_primary_key`]; bucket sharding additionally
requires it to be the single column being bucketed.
By default the MemWAL maintains every index on the table, resolved
By default the RTE maintains every index on the table, resolved
here — a snapshot, so an index created afterwards needs the spec unset
and set again. This fails if one cannot be maintained; name the set
with ``with_maintained_indexes`` to install anyway. That pins an exact
@@ -5650,30 +5672,30 @@ class AsyncTable:
Parameters
----------
spec : MemWalWriteSpec
spec : RTEWriteSpec
The sharding spec to install.
Examples
--------
>>> from lancedb import MemWalWriteSpec
>>> from lancedb import RTEWriteSpec
>>> # table.set_unenforced_primary_key("id")
>>> # table.enable_memwal(MemWalWriteSpec.bucket("id", 16))
>>> # table.enable_rte(RTEWriteSpec.bucket("id", 16))
"""
await self._inner.enable_memwal(spec)
await self._inner.enable_rte(spec)
async def disable_memwal(self) -> None:
"""Remove the MemWalWriteSpec from this table.
async def disable_rte(self) -> None:
"""Remove the RTEWriteSpec from this table.
Reverts to the standard `merge_insert` write path. Errors if no spec
is currently set.
"""
await self._inner.disable_memwal()
await self._inner.disable_rte()
async def get_memwal_spec(self) -> Optional["MemWalWriteSpec"]:
"""Read the MemWalWriteSpec currently installed on this table.
async def get_rte_spec(self) -> Optional["RTEWriteSpec"]:
"""Read the RTEWriteSpec currently installed on this table.
Returns ``None`` when the MemWAL write path is not enabled at all — no
spec has been set, or one was removed with `disable_memwal`.
Returns ``None`` when the RTE write path is not enabled at all — no
spec has been set, or one was removed with `disable_rte`.
That is a different answer from a spec whose ``maintained_indexes``
is ``None``, which is an installed spec selecting indexes
automatically.
@@ -5683,7 +5705,7 @@ class AsyncTable:
or gains later, ``[]`` maintains none, and a non-empty list maintains
exactly those. All three round-trip.
"""
return await self._inner.get_memwal_spec()
return await self._inner.get_rte_spec()
# WAL-PK-FUSION: delete both hooks.
def _hybrid_pk_fusion_learned(self) -> bool:
@@ -5694,12 +5716,12 @@ class AsyncTable:
"""See [`Table._note_hybrid_pk_fusion`][lancedb.table.Table]."""
self._inner.note_hybrid_pk_fusion()
async def checkpoint_memwal(self) -> None:
"""Converge this table's MemWAL write path into its base table.
async def checkpoint_rte(self) -> None:
"""Converge this table's RTE write path into its base table.
One flush, freezing every memtable into an SSTable, then compaction triggers
until every SSTable that existed at that moment has reached base.
The loop runs client-side, reading progress from ``get_memwal_stats``.
The loop runs client-side, reading progress from ``get_rte_stats``.
Best-effort: SSTables created *while* it runs are deliberately not
waited on, which is what lets it terminate on a table taking writes.
@@ -5714,56 +5736,56 @@ class AsyncTable:
``asyncio.wait_for`` for a wall-clock bound; abandoning it partway
costs nothing.
"""
await self._inner.checkpoint_memwal()
await self._inner.checkpoint_rte()
async def flush_memwal(self) -> None:
async def flush_rte(self) -> None:
"""Freeze every table shard's active memtable into an SSTable.
Does not touch the base table — compacting SSTables into base is
`compact_memwal`. On a node that has not claimed this table, this claims
`compact_rte`. On a node that has not claimed this table, this claims
it and replays its WAL log first.
"""
await self._inner.flush_memwal()
await self._inner.flush_rte()
async def compact_memwal(self) -> None:
async def compact_rte(self) -> None:
"""Trigger a background SSTable compaction pass per table shard.
Returns once the passes are dispatched, not once they finish: watch
``get_memwal_stats`` for progress, or use ``checkpoint_memwal`` to loop
``get_rte_stats`` for progress, or use ``checkpoint_rte`` to loop
until the current SSTables have reached base.
"""
await self._inner.compact_memwal()
await self._inner.compact_rte()
async def get_memwal_stats(
async def get_rte_stats(
self, *, include_sstable_rows: bool = False
) -> Optional[dict]:
"""Read live per-table-shard MemWAL state.
"""Read live per-table-shard RTE state.
Answers "how far behind is my fresh tier", "which table shard is hot", and
"why is my fresh-tier vector search brute-force". Mutates no table
state, though on a node that has not claimed this table it claims it,
exactly as a read would.
Returns ``None`` only when the MemWAL write path is not enabled.
Returns ``None`` only when the RTE write path is not enabled.
Parameters
----------
include_sstable_rows
Report a row count per SSTable. Off by default: each count
opens an uncached Lance dataset, and ``checkpoint_memwal`` polls this
opens an uncached Lance dataset, and ``checkpoint_rte`` polls this
needing only generation numbers.
"""
return await self._inner.get_memwal_stats(include_sstable_rows)
return await self._inner.get_rte_stats(include_sstable_rows)
async def close_memwal_writers(self) -> None:
"""Drain and close any cached MemWAL shard writers for this table.
async def close_rte_writers(self) -> None:
"""Drain and close any cached RTE shard writers for this table.
When a MemWAL write spec is installed, `merge_insert` opens MemWAL shard
When an RTE write spec is installed, `merge_insert` opens RTE shard
writers and caches them for reuse across calls. This closes them,
flushing pending data; writers reopen lazily on the next
`merge_insert`. It is a no-op when no writers are cached.
"""
await self._inner.close_memwal_writers()
await self._inner.close_rte_writers()
@property
def name(self) -> str:
@@ -6234,8 +6256,8 @@ class AsyncTable:
- pyarrow.Table or pyarrow.RecordBatch
mode: str
The mode to use when writing the data. Valid values are
"append" and "overwrite". Overwrite infers a replacement schema
but preserves existing blob v2 fields present in the new data.
"append" and "overwrite". Overwrite adopts explicitly declared blob
layouts, while inferred bytes retain the existing blob v2 fields.
on_bad_vectors: str, default "error"
What to do if any of the vectors are not the same size or contains NaNs.
One of "error", "drop", "fill", "null".
@@ -6652,8 +6674,8 @@ class AsyncTable:
take_query = self.take_offsets(query.take_offsets)
if query.columns:
take_query = take_query.select(query.columns)
if query.use_memwal is not None:
take_query = take_query.use_memwal(query.use_memwal)
if query.use_rte is not None:
take_query = take_query.use_rte(query.use_rte)
if query.with_row_id:
take_query = take_query.with_row_id()
return take_query
@@ -6669,8 +6691,8 @@ class AsyncTable:
async_query = async_query.where(query.filter)
if query.fast_search:
async_query = async_query.fast_search()
if query.use_memwal is not None:
async_query = async_query.use_memwal(query.use_memwal)
if query.use_rte is not None:
async_query = async_query.use_rte(query.use_rte)
if query.with_row_id:
async_query = async_query.with_row_id()
if query.order_by:
@@ -6796,7 +6818,7 @@ class AsyncTable:
when_not_matched_by_source_condition_expr=merge._when_not_matched_by_source_condition_expr,
timeout=merge._timeout,
use_index=merge._use_index,
use_memwal=merge._use_memwal,
use_rte=merge._use_rte,
validate_single_shard=merge._validate_single_shard,
),
)
+230
View File
@@ -63,6 +63,25 @@ def _blob_array(name, values):
return pa.ExtensionArray.from_storage(blob_type, storage)
def _blob_layout_data(rows, with_ranges, *, as_extension=True):
field = lancedb.blob("image")
storage_type = field.type.storage_type
if not with_ranges or not as_extension:
storage_type = pa.struct(list(storage_type)[: (4 if with_ranges else 2)])
field = pa.field(
"image",
storage_type,
metadata={"ARROW:extension:name": "lance.blob.v2"},
)
images = pa.array([row["image"] for row in rows], type=storage_type)
if isinstance(field.type, pa.ExtensionType):
images = pa.ExtensionArray.from_storage(field.type, images)
return pa.Table.from_arrays(
[pa.array([row["id"] for row in rows], type=pa.int64()), images],
schema=pa.schema([pa.field("id", pa.int64()), field]),
)
def _row_ids_by_id(table):
hits = table.search().with_row_id(True).limit(1000).to_arrow()
assert "_rowid" in hits.column_names
@@ -998,6 +1017,133 @@ def test_merge_insert_writes_python_bytes():
assert blobs.to_pylist() == [b"updated", b"inserted"]
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("input_type", ["extension", "struct", "reader"])
@pytest.mark.parametrize("target_with_ranges", [False, True])
async def test_merge_insert_normalizes_blob_layouts(
use_async, input_type, target_with_ranges
):
data = _blob_layout_data(
[{"id": 1, "image": {"data": b"before"}}], target_with_ranges
)
table = lancedb.connect("memory:///").create_table("merge_layouts", data=data)
original_schema = table.schema
updates = _blob_layout_data(
[
{"id": 1, "image": {"data": b"updated"}},
{"id": 2, "image": {"data": b"inserted"}},
{"id": 3, "image": None},
],
not target_with_ranges,
as_extension=input_type != "struct",
)
if input_type == "reader":
updates = updates.to_reader(max_chunksize=1)
writer = table._table if use_async else table
result = (
writer.merge_insert("id")
.when_matched_update_all()
.when_not_matched_insert_all()
.execute(updates)
)
if use_async:
result = await result
assert result.num_updated_rows == 1
assert result.num_inserted_rows == 2
assert table.schema == original_schema
by_id = _row_ids_by_id(table)
assert table.fetch_blobs("image", [by_id[i] for i in (1, 2, 3)]).to_pylist() == [
b"updated",
b"inserted",
None,
]
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize(
"byte_range", [{"position": 0}, {"size": 6}, {"position": 4, "size": 6}]
)
async def test_merge_insert_legacy_blob_rejects_ranges_atomically(
use_async, byte_range
):
data = _blob_layout_data([{"id": 1, "image": {"data": b"before"}}], False)
table = lancedb.connect("memory:///").create_table("reject_ranges", data=data)
original_schema, original_version = table.schema, table.version
updates = _blob_layout_data(
[
{"id": 1, "image": {"data": b"updated"}},
{"id": 2, "image": {"uri": "file:///payload", **byte_range}},
],
True,
).to_reader(max_chunksize=1)
writer = table._table if use_async else table
with pytest.raises(RuntimeError, match="cannot store byte ranges"):
result = (
writer.merge_insert("id")
.when_matched_update_all()
.when_not_matched_insert_all()
.execute(updates)
)
if use_async:
await result
assert table.version == original_version
assert table.schema == original_schema
assert table.count_rows() == 1
assert table.fetch_blobs("image", [_row_ids_by_id(table)[1]]).to_pylist() == [
b"before"
]
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("input_type", ["extension", "struct", "reader"])
@pytest.mark.parametrize("source_with_ranges", [False, True])
async def test_add_overwrite_adopts_explicit_blob_layout(
tmp_path, use_async, input_type, source_with_ranges
):
data = _blob_layout_data(
[{"id": 1, "image": {"data": b"before"}}], not source_with_ranges
)
table = lancedb.connect("memory:///").create_table("overwrite_layouts", data=data)
payload = b"0123456789abcdef"
path = tmp_path / "payload.bin"
path.write_bytes(payload)
value = (
{"uri": path.as_uri(), "position": 4, "size": 6}
if source_with_ranges
else {"data": b"updated"}
)
updates = _blob_layout_data(
[{"id": 2, "image": value}, {"id": 3, "image": None}],
source_with_ranges,
as_extension=input_type != "struct",
)
if input_type == "reader":
updates = updates.to_reader(max_chunksize=1)
writer = table._table if use_async else table
result = writer.add(
updates, mode="overwrite", allow_external_blob_outside_bases=True
)
if use_async:
await result
blob_type = table.schema.field("image").type
if isinstance(blob_type, pa.ExtensionType):
blob_type = blob_type.storage_type
assert blob_type.num_fields == (4 if source_with_ranges else 2)
assert table.blob_columns() == ["image"]
assert table.count_rows() == 2
by_id = _row_ids_by_id(table)
assert table.fetch_blobs("image", [by_id[2], by_id[3]]).to_pylist() == [
payload[4:10] if source_with_ranges else b"updated",
None,
]
@pytest.mark.parametrize("write", ["merge_insert", "create_table"])
def test_uri_string_blob_write_reaches_external_base_validation(tmp_path, write):
blob_path = tmp_path / "payload.bin"
@@ -2305,6 +2451,90 @@ def test_add_external_uri_string_round_trips_with_flag(tmp_path):
assert blobs[0].as_py() == payload
@pytest.mark.parametrize("as_pandas", [False, True])
@pytest.mark.parametrize("as_blob", [False, True])
@pytest.mark.parametrize("minimal_schema", [False, True])
@pytest.mark.parametrize("position", [0, 4])
def test_add_external_blob_range_matches_storage_schema(
tmp_path, as_pandas, as_blob, minimal_schema, position
):
payload = b"0123456789abcdef"
blob_path = tmp_path / "payload.bin"
blob_path.write_bytes(payload)
field = (
pa.field("image", pa.large_binary(), metadata={"lance-encoding:blob": "true"})
if minimal_schema
else lancedb.blob("image")
)
db = lancedb.connect(tmp_path / "db")
table = db.create_table(
"ranges", schema=pa.schema([pa.field("id", pa.int64()), field])
)
value = (
Blob.from_uri(blob_path.as_uri(), position=position, size=6)
if as_blob
else {"uri": blob_path.as_uri(), "position": position, "size": 6}
)
rows = [{"id": 1, "image": b"inline"}, {"id": 2, "image": value}]
data = pd.DataFrame(rows) if as_pandas else rows
if minimal_schema:
with pytest.raises(ValueError, match="Blob column 'image'.*storage schema"):
table.add(data, allow_external_blob_outside_bases=True)
assert table.count_rows() == 0
else:
table.add(data, allow_external_blob_outside_bases=True)
by_id = _row_ids_by_id(table)
assert table.fetch_blobs("image", [by_id[1], by_id[2]]).to_pylist() == [
b"inline",
payload[position : position + 6],
]
@pytest.mark.parametrize("as_pandas", [False, True])
def test_add_minimal_blob_schema_preserves_whole_file_and_inline_values(
tmp_path, as_pandas
):
payload = b"whole-file"
blob_path = tmp_path / "payload.bin"
blob_path.write_bytes(payload)
uri = blob_path.as_uri()
schema = pa.schema(
[
pa.field("id", pa.int64()),
pa.field(
"image", pa.large_binary(), metadata={"lance-encoding:blob": "true"}
),
]
)
table = lancedb.connect(tmp_path / "db").create_table("minimal", schema=schema)
values = [
None,
b"inline",
Blob.from_bytes(b"from-blob"),
uri,
Blob.from_uri(uri),
{"uri": uri, "position": None, "size": None},
]
rows = [{"id": i, "image": value} for i, value in enumerate(values)]
table.add(
pd.DataFrame(rows) if as_pandas else rows,
allow_external_blob_outside_bases=True,
)
by_id = _row_ids_by_id(table)
assert table.fetch_blobs(
"image", [by_id[i] for i in range(len(values))]
).to_pylist() == [
None,
b"inline",
b"from-blob",
payload,
payload,
payload,
]
@pytest.mark.parametrize("as_pandas", [False, True])
def test_add_bytes_and_uri_in_one_batch_preserves_external_uri(tmp_path, as_pandas):
payload = b"external-payload"
@@ -2036,6 +2036,92 @@ def test_transform_queue_depth_bounds_cooked_rows(lance_table):
)
def test_raw_queue_depth_bounded_when_transform_is_the_bottleneck(tmp_path):
"""io_queue_depth bounds raw_batches even when transform is the bottleneck.
Regression test for a bug where Stage 1 (I/O) applied no backpressure at
all: _fill_io only capped io_pending (in-flight fetches) at
io_queue_depth, but every completed fetch was drained into raw_batches
unconditionally (_drain_io) and immediately replaced by a new fetch
(_fill_io again), so I/O kept racing arbitrarily far ahead of a slow
transform stage -- raw_batches had no cap of its own.
transform_queue_depth did not help either: its admission check in
_try_submit_tx only counts cooked rows and in-flight transform futures
(`len(cooked[i]) + len(tx_pending[i]) * batch_size`); it never looks at
len(raw_batches[i]), so it cannot throttle Stage 1.
The fix makes _fill_io count both io_pending and raw_batches against the
io_queue_depth budget, so raw_queue_depth should never exceed
io_queue_depth * read_batch_size per split, regardless of how slow the
transform stage is.
I/O is forced to run synchronously (same technique as
test_transform_queue_depth_does_not_admit_at_capacity_minus_one) so the
assertion is deterministic regardless of machine load, rather than
relying on a real-time race between I/O and transform threads. Only the
transform runs on a real background thread (with a short sleep) so the
consumer genuinely blocks waiting for it, letting Stage 1 attempt to run
ahead between blocking waits.
"""
import concurrent.futures as cf
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import patch
import pyarrow as pa
import time
n_rows = 200
read_batch_size = 10
io_queue_depth = 2
transform_queue_depth = 2
max_allowed = io_queue_depth * read_batch_size
db = lancedb.connect(tmp_path)
table = db.create_table("t", pa.table({"id": list(range(n_rows))}))
def slow_transform(batch: pa.RecordBatch) -> list:
time.sleep(0.005)
return batch.column("id").to_pylist()
original_submit = ThreadPoolExecutor.submit
def synchronous_io_submit(self, fn, *args, **kwargs):
if getattr(fn, "__name__", "") == "_io_call":
fut = cf.Future()
try:
fut.set_result(fn(*args, **kwargs))
except Exception as exc:
fut.set_exception(exc)
return fut
return original_submit(self, fn, *args, **kwargs)
with patch.object(ThreadPoolExecutor, "submit", synchronous_io_submit):
ds = StreamingDataset(
table,
num_splits=1,
shuffle_seed=SHUFFLE_SEED,
read_batch_size=read_batch_size,
io_queue_depth=io_queue_depth,
transform_queue_depth=transform_queue_depth,
transform_parallelism=1,
transform=slow_transform,
)
peak_raw = 0
ids = []
for item in ds:
ids.append(item) # slow_transform yields bare ids, not dicts
if ds.raw_queue_depth > peak_raw:
peak_raw = ds.raw_queue_depth
assert peak_raw <= max_allowed, (
f"raw_queue_depth peaked at {peak_raw}, expected <= {max_allowed} "
"(io_queue_depth * read_batch_size)"
)
# The backpressure fix must not come at the cost of dropping or
# duplicating rows.
assert sorted(ids) == list(range(n_rows))
def test_transform_queue_depth_does_not_admit_at_capacity_minus_one(tmp_path):
"""Admission requires a full read_batch_size of free space, not just one slot.
@@ -3173,6 +3259,22 @@ def test_doc_example_checkpoint(lance_table):
)
def test_streaming_dataset_over_table_with_schema_metadata(tmp_path):
"""StreamingDataset builds its permutation at construction; base table schema
metadata (e.g. Hugging Face's `source-column-name-map`) must not break it."""
schema = pa.schema(
[pa.field("id", pa.int64())],
metadata={b"source-column-name-map": b"{}", b"huggingface": b"{}"},
)
table = lancedb.connect(tmp_path).create_table(
"data", pa.table({"id": list(range(NUM_ROWS))}, schema=schema)
)
ds = StreamingDataset(table, num_splits=NUM_SPLITS, shuffle_seed=SHUFFLE_SEED)
assert sorted(s["id"] for s in ds) == list(range(NUM_ROWS))
# ---------------------------------------------------------------------------
# Remote tables (LanceDB Cloud / Enterprise)
# ---------------------------------------------------------------------------
@@ -0,0 +1,291 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""
Restoring an embedding function from table metadata must not let the table grant
itself trust: remote model code and custom endpoints need approval from the
process that opens the table.
"""
import json
from unittest.mock import patch
import lancedb
import pyarrow as pa
import pytest
from lancedb.embeddings import (
EmbeddingFunctionConfig,
OllamaEmbeddings,
OpenAIEmbeddings,
SentenceTransformerEmbeddings,
UntrustedEmbeddingConfigError,
get_registry,
)
from lancedb.embeddings.base import UntrustedEmbeddingFunction
@pytest.fixture(autouse=True)
def fresh_trust(monkeypatch):
registry = get_registry()
monkeypatch.setattr(registry, "_allowed_remote_code", set())
monkeypatch.setattr(registry, "_allowed_endpoints", set())
def metadata_for(name: str, model: dict) -> dict:
entry = {
"name": name,
"model": model,
"source_column": "text",
"vector_column": "vector",
}
return {b"embedding_functions": json.dumps([entry]).encode()}
def restore(name: str, model: dict):
return get_registry().parse_functions(metadata_for(name, model))["vector"].function
def test_sentence_transformers_does_not_trust_remote_code_by_default():
assert SentenceTransformerEmbeddings.create().trust_remote_code is False
@pytest.mark.parametrize("name", ["sentence-transformers", "huggingface", "colbert"])
def test_restored_remote_code_requires_allowlist(name):
model = {"name": "attacker/model", "trust_remote_code": True}
func = restore(name, model)
with pytest.raises(UntrustedEmbeddingConfigError, match="attacker/model"):
func.compute_query_embeddings("hello")
with pytest.raises(UntrustedEmbeddingConfigError):
func.ndims()
def test_blocked_function_does_not_construct_model():
# huggingface loads its model in __init__, so the check has to happen first.
with patch(
"lancedb.embeddings.transformers.attempt_import_or_raise"
) as attempt_import:
restore("huggingface", {"name": "attacker/model", "trust_remote_code": True})
attempt_import.assert_not_called()
def test_allowlisted_model_restores_with_remote_code():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers", {"name": "org/model", "trust_remote_code": True}
)
assert isinstance(func, SentenceTransformerEmbeddings)
assert func.trust_remote_code is True
def test_allowlist_does_not_cover_other_models():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers", {"name": "org/other", "trust_remote_code": True}
)
with pytest.raises(UntrustedEmbeddingConfigError, match="org/other"):
func.compute_query_embeddings("hello")
def test_pinned_revision_must_match():
get_registry().allow_remote_code("org/model", revision="abc123")
pinned = restore(
"sentence-transformers",
{"name": "org/model", "revision": "abc123", "trust_remote_code": True},
)
assert pinned.trust_remote_code is True
for model in [
{"name": "org/model", "revision": "def456", "trust_remote_code": True},
{"name": "org/model", "trust_remote_code": True},
]:
func = restore("sentence-transformers", model)
with pytest.raises(UntrustedEmbeddingConfigError, match="org/model"):
func.compute_query_embeddings("hello")
def test_restored_trust_flag_follows_allowlist_when_absent():
# Tables written before the default changed don't record trust_remote_code.
assert restore(
"sentence-transformers", {"name": "org/model"}
).trust_remote_code is (False)
get_registry().allow_remote_code("org/model")
assert restore("sentence-transformers", {"name": "org/model"}).trust_remote_code
def test_created_function_records_trust_flag():
# Otherwise a new table using the default would look like a pre-fix table,
# where a missing flag meant True.
conf = EmbeddingFunctionConfig(
source_column="text",
vector_column="vector",
function=SentenceTransformerEmbeddings.create(name="org/model"),
)
assert get_registry().function_to_metadata(conf)["model"] == {
"name": "org/model",
"trust_remote_code": False,
}
get_registry().allow_remote_code("org/model")
metadata = get_registry().get_table_metadata([conf])
restored = get_registry().parse_functions(metadata)["vector"].function
assert restored.trust_remote_code is False
def test_huggingface_missing_flag_keeps_false_default():
# huggingface always defaulted to False, so approval must not flip it.
get_registry().allow_remote_code("org/model")
with patch("lancedb.embeddings.transformers.attempt_import_or_raise"):
func = restore("huggingface", {"name": "org/model"})
assert func.trust_remote_code is False
def test_restored_explicit_false_stays_false():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers", {"name": "org/model", "trust_remote_code": False}
)
assert func.trust_remote_code is False
def test_create_trusts_its_own_config(tmp_path):
func = SentenceTransformerEmbeddings.create(
name="org/model", trust_remote_code=True
)
db = lancedb.connect(tmp_path)
schema = pa.schema(
[pa.field("text", pa.string()), pa.field("vector", pa.list_(pa.float32(), 4))]
)
db.create_table(
"t",
schema=schema,
embedding_functions=[
EmbeddingFunctionConfig(
source_column="text", vector_column="vector", function=func
)
],
)
restored = db.open_table("t").embedding_functions["vector"].function
assert isinstance(restored, SentenceTransformerEmbeddings)
assert restored.trust_remote_code is True
def test_create_trusts_only_the_revision_it_selected():
# Anyone can push a PR ref to a Hugging Face repo, so trusting a model's
# default branch must not extend to its other revisions.
SentenceTransformerEmbeddings.create(name="org/model", trust_remote_code=True)
func = restore(
"sentence-transformers",
{"name": "org/model", "revision": "refs/pr/123", "trust_remote_code": True},
)
assert isinstance(func, UntrustedEmbeddingFunction)
same = restore(
"sentence-transformers", {"name": "org/model", "trust_remote_code": True}
)
assert same.trust_remote_code is True
def test_explicit_unpinned_approval_covers_any_revision():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers",
{"name": "org/model", "revision": "abc123", "trust_remote_code": True},
)
assert func.trust_remote_code is True
def test_missing_flag_does_not_inherit_unapproved_trust_default():
class LegacySentenceTransformer(SentenceTransformerEmbeddings):
trust_remote_code: bool = True
registry = get_registry()
registry.register("legacy-trust-default")(LegacySentenceTransformer)
try:
func = restore("legacy-trust-default", {"name": "attacker/model"})
assert not func.trust_remote_code
finally:
del registry._functions["legacy-trust-default"]
def test_trusting_one_model_does_not_trust_a_swapped_name():
SentenceTransformerEmbeddings.create(name="org/model", trust_remote_code=True)
func = restore(
"sentence-transformers", {"name": "attacker/model", "trust_remote_code": True}
)
with pytest.raises(UntrustedEmbeddingConfigError):
func.compute_query_embeddings("hello")
ATTACKER_URL = "https://attacker.example/v1"
@pytest.mark.parametrize(
"name,model",
[
("openai", {"base_url": ATTACKER_URL}),
("ollama", {"host": ATTACKER_URL}),
("ollama", {"ollama_client_kwargs": {"proxy": ATTACKER_URL}}),
("watsonx", {"url": ATTACKER_URL}),
],
)
def test_restored_custom_endpoint_requires_allowlist(name, model):
func = restore(name, model)
with pytest.raises(UntrustedEmbeddingConfigError, match="attacker.example"):
func.compute_query_embeddings("hello")
get_registry().allow_endpoint(ATTACKER_URL)
assert isinstance(restore(name, model), get_registry().get(name))
def test_default_endpoints_need_no_allowlist():
assert isinstance(restore("openai", {}), OpenAIEmbeddings)
assert isinstance(
restore("ollama", {"host": "http://localhost:11434"}), OllamaEmbeddings
)
def test_endpoint_allowlist_ignores_trailing_slash():
get_registry().allow_endpoint("http://my-host:11434/")
assert isinstance(
restore("ollama", {"host": "http://my-host:11434"}), OllamaEmbeddings
)
def test_create_trusts_its_own_endpoint():
OpenAIEmbeddings.create(base_url="http://my-proxy/v1")
assert isinstance(
restore("openai", {"base_url": "http://my-proxy/v1"}), OpenAIEmbeddings
)
def test_blocked_function_round_trips_metadata():
model = {"name": "attacker/model", "trust_remote_code": True}
func = restore("sentence-transformers", model)
conf = EmbeddingFunctionConfig(
source_column="text", vector_column="vector", function=func
)
entry = get_registry().function_to_metadata(conf)
assert entry["name"] == "sentence-transformers"
assert entry["model"] == model
def test_table_with_blocked_function_still_opens(tmp_path):
db = lancedb.connect(tmp_path)
schema = pa.schema(
[pa.field("text", pa.string()), pa.field("vector", pa.list_(pa.float32(), 2))],
metadata=metadata_for(
"sentence-transformers",
{"name": "attacker/model", "trust_remote_code": True},
),
)
db.create_table("t", schema=schema)
tbl = db.open_table("t")
tbl.add([{"text": "hi", "vector": [1.0, 2.0]}])
assert tbl.search([1.0, 2.0]).limit(1).to_list()[0]["text"] == "hi"
with pytest.raises(UntrustedEmbeddingConfigError, match="allow_remote_code"):
tbl.search("hi").limit(1).to_list()
@@ -622,7 +622,6 @@ def test_canonical_arrow_type_prefers_the_compact_grammar():
for outside in [
pa.timestamp("us"),
pa.decimal128(10, 2),
pa.large_binary(),
pa.binary(4),
pa.duration("s"),
pa.list_(pa.float32(), 0),
@@ -1280,6 +1279,28 @@ def test_explicit_large_utf8_schemas_use_the_canonical_function_name():
}
def test_large_binary_schemas_use_the_canonical_function_name():
payload = pa.field("data", pa.large_binary(), nullable=True)
@udf(
input_schema=pa.schema([payload]),
output_schema=pa.field("result", pa.large_binary(), nullable=False),
)
def preserve(data):
return data
signature = preserve.registration_request.signature
assert signature.inputs[0].arrow_type == "large_binary"
assert signature.output.arrow_type == "large_binary"
cells = pa.large_list(pa.field("item", pa.struct([payload]), nullable=True))
assert json.loads(_canonical_arrow_type(cells))["fields"][0]["type"] == {
"type": "struct",
"fields": [
{"name": "data", "nullable": True, "type": {"type": "large_binary"}}
],
}
def test_nested_struct_output_uses_canonical_exact_json():
token = pa.struct(
[
+50 -56
View File
@@ -1,14 +1,14 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Tests for the MemWAL LSM ``merge_insert`` dispatch."""
"""Tests for the RTE LSM ``merge_insert`` dispatch."""
from datetime import timedelta
import lancedb
import pyarrow as pa
import pytest
from lancedb._lancedb import MemWalWriteSpec
from lancedb._lancedb import RTEWriteSpec
from lancedb.index import FTS, IvfPq
SCHEMA = pa.schema(
@@ -49,12 +49,12 @@ def _region_reader(rows):
def _bucket_table(tmp_path):
"""A table with ``id`` as the primary key and a single-bucket MemWAL spec."""
"""A table with ``id`` as the primary key and a single-bucket RTE spec."""
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id")
# num_buckets = 1: every row routes to the single bucket.
table.enable_memwal(MemWalWriteSpec.bucket("id", 1))
table.enable_rte(RTEWriteSpec.bucket("id", 1))
return table
@@ -67,7 +67,7 @@ def test_lsm_merge_insert_bucket(tmp_path):
.when_not_matched_insert_all()
.execute(_reader([3, 4, 5]))
)
# LSM path: rows go to the MemWAL, so only num_rows is populated.
# LSM path: rows go to the RTE, so only num_rows is populated.
assert result.num_rows == 3
assert result.version == 0
assert result.num_inserted_rows == 0
@@ -78,7 +78,7 @@ def test_lsm_merge_insert_unsharded(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.unsharded())
table.enable_rte(RTEWriteSpec.unsharded())
result = (
table.merge_insert("id")
.when_matched_update_all()
@@ -92,7 +92,7 @@ def test_lsm_merge_insert_identity(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _region_reader([(1, "us"), (2, "us")]))
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.identity("region"))
table.enable_rte(RTEWriteSpec.identity("region"))
# All rows share one identity value, so they route to one shard.
result = (
table.merge_insert([])
@@ -103,31 +103,31 @@ def test_lsm_merge_insert_identity(tmp_path):
assert result.num_rows == 2
def test_lsm_merge_insert_use_memwal_false(tmp_path):
def test_lsm_merge_insert_use_rte_false(tmp_path):
table = _bucket_table(tmp_path) # rows id = 1, 2, 3
# use_memwal(False) opts out: the standard path runs and commits even with a spec.
# use_rte(False) opts out: the standard path runs and commits even with a spec.
result = (
table.merge_insert("id")
.when_not_matched_insert_all()
.use_memwal(False)
.use_rte(False)
.execute(_reader([3, 4, 5]))
)
assert result.num_inserted_rows == 2
assert table.count_rows() == 5
def test_lsm_merge_insert_use_memwal_true_without_spec_errors(tmp_path):
# A table with a primary key but no MemWAL write spec installed.
def test_lsm_merge_insert_use_rte_true_without_spec_errors(tmp_path):
# A table with a primary key but no RTE write spec installed.
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id")
# use_memwal(True) demands MemWAL routing; without a spec it errors.
with pytest.raises(Exception, match="use_memwal"):
# use_rte(True) demands RTE routing; without a spec it errors.
with pytest.raises(Exception, match="use_rte"):
(
table.merge_insert("id")
.when_matched_update_all()
.when_not_matched_insert_all()
.use_memwal(True)
.use_rte(True)
.execute(_reader([3, 4, 5]))
)
@@ -145,7 +145,7 @@ def test_lsm_merge_insert_validate_single_shard_off(tmp_path):
def test_lsm_merge_insert_no_spec_uses_standard_path(tmp_path):
# A table with a primary key but no MemWAL write spec installed.
# A table with a primary key but no RTE write spec installed.
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id")
@@ -186,7 +186,7 @@ def test_lsm_close_writers(tmp_path):
.when_not_matched_insert_all()
.execute(_reader([7, 8]))
)
table.close_memwal_writers()
table.close_rte_writers()
# The writer reopens lazily on the next merge_insert.
result = (
table.merge_insert([])
@@ -204,14 +204,14 @@ async def test_async_lsm_merge_insert(tmp_path):
)
table = await db.create_table("t", _reader([1, 2, 3]))
await table.set_unenforced_primary_key("id")
await table.enable_memwal(MemWalWriteSpec.bucket("id", 1))
await table.enable_rte(RTEWriteSpec.bucket("id", 1))
builder = (
table.merge_insert([]).when_matched_update_all().when_not_matched_insert_all()
)
result = await builder.execute(_reader([3, 4, 5]))
assert result.num_rows == 3
await table.close_memwal_writers()
await table.close_rte_writers()
def _lsm_upsert(table, ids):
@@ -228,7 +228,7 @@ def test_lsm_read_sees_active_memtable(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3])) # base ids 1,2,3
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.unsharded())
table.enable_rte(RTEWriteSpec.unsharded())
_lsm_upsert(table, [4, 5]) # active memtable only, not committed to base
@@ -236,8 +236,8 @@ def test_lsm_read_sees_active_memtable(tmp_path):
lsm = table.search().to_arrow()
assert sorted(lsm["id"].to_pylist()) == [1, 2, 3, 4, 5]
# use_memwal(False) bypasses the MemWAL and reads the base table only.
base_only = table.search().use_memwal(False).to_arrow()
# use_rte(False) bypasses the RTE and reads the base table only.
base_only = table.search().use_rte(False).to_arrow()
assert sorted(base_only["id"].to_pylist()) == [1, 2, 3]
@@ -245,7 +245,7 @@ def test_lsm_read_dedup_newest_wins(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3])) # id 2 -> value 1
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.unsharded())
table.enable_rte(RTEWriteSpec.unsharded())
_lsm_upsert(table, [2, 3, 4]) # ids 2,3,4 -> values 0,1,2
@@ -258,31 +258,31 @@ def test_lsm_read_dedup_newest_wins(tmp_path):
def test_lsm_read_without_spec_reads_base(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id") # no MemWAL write spec
table.set_unenforced_primary_key("id") # no RTE write spec
# No spec: default read and use_memwal(False) both read the base table, no error.
# No spec: default read and use_rte(False) both read the base table, no error.
assert sorted(table.search().to_arrow()["id"].to_pylist()) == [1, 2, 3]
assert sorted(table.search().use_memwal(False).to_arrow()["id"].to_pylist()) == [
assert sorted(table.search().use_rte(False).to_arrow()["id"].to_pylist()) == [
1,
2,
3,
]
def test_lsm_read_unsupported_shape_errors_without_use_memwal_false(tmp_path):
def test_lsm_read_unsupported_shape_errors_without_use_rte_false(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.unsharded())
table.enable_rte(RTEWriteSpec.unsharded())
_lsm_upsert(table, [4])
# with_row_id is unsupported by the LSM scanner; on a MemWAL table the default
# with_row_id is unsupported by the LSM scanner; on an RTE table the default
# (auto-routed) read hard-errors instead of silently reading a stale base.
with pytest.raises(Exception):
table.search().with_row_id(True).to_arrow()
# use_memwal(False) is the escape hatch: it reads the base table only.
base = table.search().with_row_id(True).use_memwal(False).to_arrow()
# use_rte(False) is the escape hatch: it reads the base table only.
base = table.search().with_row_id(True).use_rte(False).to_arrow()
assert sorted(base["id"].to_pylist()) == [1, 2, 3]
@@ -293,7 +293,7 @@ async def test_async_lsm_read(tmp_path):
)
table = await db.create_table("t", _reader([1, 2, 3]))
await table.set_unenforced_primary_key("id")
await table.enable_memwal(MemWalWriteSpec.unsharded())
await table.enable_rte(RTEWriteSpec.unsharded())
builder = (
table.merge_insert([]).when_matched_update_all().when_not_matched_insert_all()
@@ -331,7 +331,7 @@ def _vector_reader(rows):
def _vector_table(tmp_path):
"""Base table whose vector column is indexed so its rows are visible to the LSM
vector scanner (the base arm uses ``fast_search`` — indexed data only), plus an
unsharded MemWAL spec that maintains that index for the memtable.
unsharded RTE spec that maintains that index for the memtable.
Rows 1,2 are category ``a``, row 3 is ``b``, and 4..60 are filler ``c`` that
give the tiny IVF index enough data to train.
@@ -353,9 +353,7 @@ def _vector_table(tmp_path):
"vector", config=IvfPq(num_partitions=1, num_sub_vectors=2, num_bits=4)
)
index_name = table.list_indices()[0].name
table.enable_memwal(
MemWalWriteSpec.unsharded().with_maintained_indexes([index_name])
)
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes([index_name]))
return table
@@ -379,9 +377,9 @@ def test_lsm_read_vector_sees_memtable(tmp_path):
assert {1, 2, 3} <= ids # indexed base rows
assert 1000 in ids # in-flight memtable row
# use_memwal(False) bypasses the MemWAL, so the in-flight row is not visible.
# use_rte(False) bypasses the RTE, so the in-flight row is not visible.
base_ids = set(
table.search(query).use_memwal(False).limit(100).to_arrow()["id"].to_pylist()
table.search(query).use_rte(False).limit(100).to_arrow()["id"].to_pylist()
)
assert {1, 2, 3} <= base_ids
assert 1000 not in base_ids
@@ -453,9 +451,7 @@ def test_lsm_read_fts_sees_memtable(tmp_path):
# Native FTS index (tantivy is not compatible with the LSM memtable index).
table.create_index("text", config=FTS())
index_name = table.list_indices()[0].name
table.enable_memwal(
MemWalWriteSpec.unsharded().with_maintained_indexes([index_name])
)
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes([index_name]))
# in-flight doc 4 lands in the memtable's maintained FTS index.
(
@@ -498,8 +494,8 @@ def test_lsm_read_vector_unsupported_knobs_error(tmp_path):
with pytest.raises(Exception, match="use_index"):
table.search(query).bypass_vector_index().to_arrow()
# use_memwal(False) is the escape hatch: the base-only standard path honors them.
base = table.search(query).distance_range(0.0, 100.0).use_memwal(False).to_arrow()
# use_rte(False) is the escape hatch: the base-only standard path honors them.
base = table.search(query).distance_range(0.0, 100.0).use_rte(False).to_arrow()
assert 1000 not in set(base["id"].to_pylist())
@@ -544,7 +540,7 @@ def test_lsm_read_fts_unmaintained_index_errors(tmp_path):
table.create_index("text", config=FTS())
# No maintained indexes: the active memtable FTS arm cannot serve un-compacted
# docs, so the search would silently omit them — reject instead.
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([]))
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes([]))
with pytest.raises(Exception, match="maintained"):
table.search("fox", query_type="fts", fts_columns="text").to_arrow()
@@ -553,7 +549,7 @@ def test_lsm_read_time_travel_errors(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.unsharded())
table.enable_rte(RTEWriteSpec.unsharded())
pinned = table.version
table.add(_reader([4, 5])) # standard add commits a newer version
table.checkout(pinned) # detached head at the historical version
@@ -562,8 +558,8 @@ def test_lsm_read_time_travel_errors(tmp_path):
# historical version is rejected.
with pytest.raises(Exception, match="time-travel"):
table.search().to_arrow()
# use_memwal(False) reads the base table at the pinned version.
base = table.search().use_memwal(False).to_arrow()
# use_rte(False) reads the base table at the pinned version.
base = table.search().use_rte(False).to_arrow()
assert sorted(base["id"].to_pylist()) == [1, 2, 3]
@@ -571,14 +567,14 @@ def test_lsm_read_take_row_ids_errors(tmp_path):
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(seconds=0))
table = db.create_table("t", _reader([1, 2, 3]))
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.unsharded())
table.enable_rte(RTEWriteSpec.unsharded())
_lsm_upsert(table, [4])
# take-by-row-id auto-routes through the LSM scanner, which has no stable _rowid,
# so it hard-errors instead of failing with an opaque column-not-found error.
with pytest.raises(Exception, match="row id"):
table.take_row_ids([0, 1]).to_arrow()
# use_memwal(False) is the escape hatch: it reads the base table.
base = table.take_row_ids([0, 1]).use_memwal(False).to_arrow()
# use_rte(False) is the escape hatch: it reads the base table.
base = table.take_row_ids([0, 1]).use_rte(False).to_arrow()
assert base.num_rows == 2
@@ -588,9 +584,7 @@ def test_lsm_read_fts_postfilter_errors(tmp_path):
table.set_unenforced_primary_key("id")
table.create_index("text", config=FTS())
index_name = table.list_indices()[0].name
table.enable_memwal(
MemWalWriteSpec.unsharded().with_maintained_indexes([index_name])
)
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes([index_name]))
# The LSM scanner always prefilters; postfilter on FTS changes result semantics,
# so it hard-errors (previously only the vector arm rejected it).
with pytest.raises(Exception, match="postfilter"):
@@ -608,7 +602,7 @@ def test_lsm_read_fts_multiple_same_type_indexes_errors(tmp_path):
table.set_unenforced_primary_key("id")
table.create_index("text", config=FTS(), name="fts_a")
table.create_index("text", config=FTS(), name="fts_b", replace=False)
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes(["fts_a"]))
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes(["fts_a"]))
# Two FTS indexes on the column: the base planner's chosen index is ambiguous, so
# the scanner can't pick a catch-up watermark and rejects rather than risk
# dropping rows the actually-used index has not caught up to.
@@ -629,7 +623,7 @@ def test_lsm_read_vector_unmaintained_index_errors(tmp_path):
)
# Spec with NO maintained indexes: the base vector index's catch-up is untracked,
# so the scanner rejects rather than risk dropping compacted-but-unindexed rows.
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([]))
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes([]))
with pytest.raises(Exception, match="maintained"):
table.search([1.0] * VECTOR_DIM).to_arrow()
@@ -642,7 +636,7 @@ def test_lsm_read_fts_optimized_index_not_rejected(tmp_path):
table.add(_fts_reader([(i, "lazy fox") for i in range(6, 11)]))
table.optimize() # may split the FTS index into multiple physical segments
name = table.list_indices()[0].name
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([name]))
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes([name]))
# Multiple physical segments of one logical index must not be miscounted as
# multiple indexes and rejected.
ids = set(
+29
View File
@@ -115,6 +115,35 @@ def test_split_random_fixed(mem_db):
assert split_ids.count(split_id) == 25
def test_permutation_over_table_with_schema_metadata(tmp_path):
"""Schema-level metadata on the base table must not break the build.
Hugging Face datasets carry keys like `huggingface` or `source-column-name-map`.
"""
schema = pa.schema(
[pa.field("x", pa.int64())],
# `split_names` collides with a permutation config key; the config must win.
metadata={b"huggingface": b"{}", b"split_names": b'["base"]'},
)
tbl = connect(tmp_path).create_table(
"t", pa.table({"x": list(range(100))}, schema=schema)
)
permutation_tbl = (
permutation_builder(tbl)
.split_random(fixed=2, seed=1, split_names=["train", "test"])
.execute()
)
permutations = Permutations(tbl, permutation_tbl)
assert permutations.split_names == ["train", "test"]
train = Permutation.from_tables(tbl, permutation_tbl, "train")
test = Permutation.from_tables(tbl, permutation_tbl, "test")
assert train.num_rows + test.num_rows == 100
train_x = [row["x"] for batch in train.iter(16) for row in batch]
test_x = [row["x"] for batch in test.iter(16) for row in batch]
assert sorted(train_x + test_x) == list(range(100))
def test_split_random_with_seed(mem_db):
"""Test that seeded random splits are reproducible."""
tbl = mem_db.create_table("test_table", pa.table({"x": range(50), "y": range(50)}))
+147 -22
View File
@@ -1623,7 +1623,6 @@ def remote_unsupported_operations():
return [
("uses_v2_manifest_paths", (), {}, "uses_v2_manifest_paths"),
("migrate_v2_manifest_paths", (), {}, "migrate_manifest_paths_v2"),
("fetch_blob_ranges", ("image", [(1, 0, 1)]), {}, "fetch_blob_ranges"),
("optimize", (), {}, "optimize"),
("create_index", ("vector",), {"config": HnswPq()}, "IVF_HNSW_PQ"),
]
@@ -1845,10 +1844,10 @@ def lsm_test_table(lsm_handler):
"""A remote table whose LSM routes are served by ``lsm_handler``.
``lsm_handler(request, route)`` is called for ``/v1/table/test/<route>/``
where route is one of flush_memwal, compact_memwal, get_memwal_stats, and is
where route is one of flush_rte, compact_rte, get_rte_stats, and is
responsible for writing the response.
"""
routes = ("flush_memwal", "compact_memwal", "get_memwal_stats")
routes = ("flush_rte", "compact_rte", "get_rte_stats")
def handler(request):
match = re.fullmatch(r"/v1/table/test/(\w+)/", request.path)
@@ -1880,7 +1879,7 @@ def send_json(request, payload, status=200):
request.wfile.write(json.dumps(payload).encode())
def test_get_memwal_stats_sync():
def test_get_rte_stats_sync():
"""The sync wrapper round-trips the server payload into a dict."""
table_shard = {
"shard_id": "b0",
@@ -1905,29 +1904,29 @@ def test_get_memwal_stats_sync():
seen_bodies = []
def lsm_handler(request, route):
assert route == "get_memwal_stats"
assert route == "get_rte_stats"
seen_bodies.append(read_json_body(request))
send_json(request, {"memwal_stats": {"table_shards": [table_shard]}})
send_json(request, {"rte_stats": {"table_shards": [table_shard]}})
with lsm_test_table(lsm_handler) as table:
assert table.get_memwal_stats() == {"table_shards": [table_shard]}
assert table.get_rte_stats() == {"table_shards": [table_shard]}
# Off by default, and forwarded when asked for.
assert seen_bodies == [{"include_sstable_rows": False}]
table.get_memwal_stats(include_sstable_rows=True)
table.get_rte_stats(include_sstable_rows=True)
assert seen_bodies[-1] == {"include_sstable_rows": True}
def test_get_memwal_stats_sync_returns_none_when_memwal_disabled():
"""A null envelope means the MemWAL write path is not enabled, not an error."""
def test_get_rte_stats_sync_returns_none_when_rte_disabled():
"""A null envelope means the RTE write path is not enabled, not an error."""
def lsm_handler(request, route):
send_json(request, {"memwal_stats": None})
send_json(request, {"rte_stats": None})
with lsm_test_table(lsm_handler) as table:
assert table.get_memwal_stats() is None
assert table.get_rte_stats() is None
def test_flush_and_compact_memwal_sync():
def test_flush_and_compact_rte_sync():
"""Both are one-shot POSTs answered 202 with no body."""
called = []
@@ -1937,12 +1936,12 @@ def test_flush_and_compact_memwal_sync():
request.end_headers()
with lsm_test_table(lsm_handler) as table:
assert table.flush_memwal() is None
assert table.compact_memwal() is None
assert called == ["flush_memwal", "compact_memwal"]
assert table.flush_rte() is None
assert table.compact_rte() is None
assert called == ["flush_rte", "compact_rte"]
def test_checkpoint_memwal_sync():
def test_checkpoint_rte_sync():
"""Freeze, read the watermark, and return once no SSTables remain.
The convergence loop itself is covered in Rust; this pins the sync
@@ -1952,17 +1951,17 @@ def test_checkpoint_memwal_sync():
def lsm_handler(request, route):
called.append(route)
if route == "get_memwal_stats":
if route == "get_rte_stats":
# An empty SSTable tier yields no target watermark, so the loop is done
# after the freeze without ever polling compaction.
send_json(request, {"memwal_stats": {"table_shards": []}})
send_json(request, {"rte_stats": {"table_shards": []}})
else:
request.send_response(202)
request.end_headers()
with lsm_test_table(lsm_handler) as table:
assert table.checkpoint_memwal() is None
assert called == ["flush_memwal", "get_memwal_stats"]
assert table.checkpoint_rte() is None
assert called == ["flush_rte", "get_rte_stats"]
@contextlib.contextmanager
@@ -3255,6 +3254,89 @@ BLOB_DESCRIBE_RESPONSE = {
}
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
@pytest.mark.parametrize("mode", ["merge", "overwrite"])
@pytest.mark.parametrize("target_with_ranges", [False, True])
async def test_remote_writes_normalize_blob_layouts(
use_async, mode, target_with_ranges
):
description = json.loads(json.dumps(BLOB_DESCRIBE_RESPONSE))
if target_with_ranges:
description["schema"]["fields"][1]["type"]["fields"].extend(
{"name": name, "type": {"type": "uint64"}, "nullable": True}
for name in ("position", "size")
)
received = []
def handler(request):
if request.path == "/v1/table/test/describe/":
response = description
else:
expected = "merge_insert" if mode == "merge" else "insert"
assert urlparse(request.path).path == f"/v1/table/test/{expected}/"
if request.headers.get("Transfer-Encoding") == "chunked":
body = bytearray()
while size := int(request.rfile.readline().split(b";")[0], 16):
body.extend(request.rfile.read(size))
assert request.rfile.read(2) == b"\r\n"
assert request.rfile.read(2) == b"\r\n"
else:
body = request.rfile.read(int(request.headers["Content-Length"]))
received.append(pa.ipc.open_stream(body).read_all())
response = {
"version": 2,
"num_updated_rows": 0,
"num_inserted_rows": 2,
"num_deleted_rows": 0,
}
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.send_header("phalanx-version", "0.5.0")
request.end_headers()
request.wfile.write(json.dumps(response).encode())
field = lancedb.blob("image")
storage_type = field.type.storage_type
if target_with_ranges:
storage_type = pa.struct(list(storage_type)[:2])
field = pa.field(
"image",
storage_type,
metadata={"ARROW:extension:name": "lance.blob.v2"},
)
images = pa.array([{"data": b"payload"}, None], type=storage_type)
if isinstance(field.type, pa.ExtensionType):
images = pa.ExtensionArray.from_storage(field.type, images)
data = pa.Table.from_arrays(
[pa.array([1, 2], type=pa.int64()), images],
schema=pa.schema([pa.field("id", pa.int64()), field]),
)
with mock_lancedb_connection(handler) as db:
remote_table = db.open_table("test")
table = remote_table._table if use_async else remote_table
if mode == "merge":
result = (
table.merge_insert("id").when_not_matched_insert_all().execute(data)
)
else:
result = table.add(data, mode="overwrite")
if use_async:
await result
assert len(received) == 1
images = received[0]["image"].combine_chunks()
if isinstance(images.type, pa.ExtensionType):
images = images.storage
with_ranges = target_with_ranges if mode == "merge" else not target_with_ranges
assert images.type.num_fields == (4 if with_ranges else 2)
expected = {"data": b"payload", "uri": None}
if with_ranges:
expected.update(position=None, size=None)
assert images.to_pylist() == [expected, None]
def blob_query_response_table():
image_field = pa.field(
"image",
@@ -3362,13 +3444,20 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
else:
assert body["with_row_id"] is True
if columns == ["image"]:
# fetch_blob_files sizes its handles from a descriptor take.
# Blob handles are sized from a descriptor take.
assert body["filter"].startswith("_rowid IN")
response_table = blob_descriptor_take_table()
row_ids = [
int(value) for value in re.findall(r"\d+", body["filter"])
]
response_table = response_table.filter(
pa.compute.is_in(response_table["_rowid"], pa.array(row_ids))
)
else:
response_table = blob_query_response_table()
request.send_response(200)
request.send_header("Content-Type", "application/vnd.apache.arrow.file")
request.send_header("x-lancedb-version", "1")
request.end_headers()
with pa.ipc.new_file(request.wfile, response_table.schema) as writer:
writer.write_table(response_table)
@@ -3400,6 +3489,42 @@ def test_remote_blob_columns_and_fetch():
assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True])
async def test_remote_fetch_blob_ranges(use_async):
with blob_remote_table() as remote_table:
table = remote_table._table if use_async else remote_table
requests = [
(30, 2, 3),
(10, 1, 3),
(20, 0, 1),
(10, 0, 2),
(40, 0, 0),
(10, 5, 0),
(30, 2, 3),
]
result = table.fetch_blob_ranges("image", requests)
if use_async:
result = await result
assert isinstance(result, pa.LargeBinaryArray)
assert result.to_pylist() == [b"mma", b"lph", None, b"al", b"", b"", b"mma"]
for invalid, message in [
([(10, 0, 1), (99, 0, 1)], "only 1 exist"),
([(10, 4, 2)], "exceeds blob size"),
([(20, 2**64 - 1, 1)], "overflows u64"),
]:
with pytest.raises(ValueError, match=message):
result = table.fetch_blob_ranges("image", invalid)
if use_async:
await result
result = table.fetch_blob_ranges("image", [])
if use_async:
result = await result
assert result.to_pylist() == []
@pytest.mark.asyncio
async def test_async_remote_to_lance_rejects_storage_access(monkeypatch):
with blob_remote_table() as remote_table:
@@ -1,8 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Tests for installing and clearing an MemWalWriteSpec via
`Table.enable_memwal` / `Table.disable_memwal`.
"""Tests for installing and clearing an RTEWriteSpec via
`Table.enable_rte` / `Table.disable_rte`.
"""
from datetime import timedelta
@@ -10,7 +10,7 @@ from datetime import timedelta
import lancedb
import pyarrow as pa
import pytest
from lancedb._lancedb import MemWalWriteSpec
from lancedb._lancedb import RTEWriteSpec
from lancedb.index import BTree
SCHEMA = pa.schema(
@@ -21,9 +21,9 @@ SCHEMA = pa.schema(
)
def test_memwal_spec_module_metadata():
assert lancedb.MemWalWriteSpec is MemWalWriteSpec
assert MemWalWriteSpec.__module__ == "lancedb._lancedb"
def test_rte_spec_module_metadata():
assert lancedb.RTEWriteSpec is RTEWriteSpec
assert RTEWriteSpec.__module__ == "lancedb._lancedb"
def _batch(ids, vs):
@@ -43,35 +43,49 @@ def _make_table(tmp_path):
return db, table
def test_enable_memwal_validates(tmp_path):
def test_enable_rte_validates(tmp_path):
_db, table = _make_table(tmp_path)
table.set_unenforced_primary_key("id")
# Out-of-range num_buckets.
with pytest.raises(Exception, match="num_buckets"):
table.enable_memwal(MemWalWriteSpec.bucket("id", 0))
table.enable_rte(RTEWriteSpec.bucket("id", 0))
with pytest.raises(Exception, match="num_buckets"):
table.enable_memwal(MemWalWriteSpec.bucket("id", 1025))
table.enable_rte(RTEWriteSpec.bucket("id", 1025))
# Happy path, then a set over the installed spec is refused.
table.enable_memwal(MemWalWriteSpec.bucket("id", 4))
table.enable_rte(RTEWriteSpec.bucket("id", 4))
with pytest.raises(Exception, match="already set"):
table.enable_memwal(MemWalWriteSpec.bucket("id", 8))
table.enable_rte(RTEWriteSpec.bucket("id", 8))
def test_disable_memwal(tmp_path):
def test_disable_rte(tmp_path):
_db, table = _make_table(tmp_path)
table.set_unenforced_primary_key("id")
# unset errors when no spec is set.
with pytest.raises(Exception, match="no MemWAL write spec"):
table.disable_memwal()
with pytest.raises(Exception, match="no RTE write spec"):
table.disable_rte()
# Install a spec, then remove it; afterwards a fresh spec can be set.
table.enable_memwal(MemWalWriteSpec.bucket("id", 4))
table.disable_memwal()
table.enable_rte(RTEWriteSpec.bucket("id", 4))
table.disable_rte()
# A second unset errors — there is no spec left to remove.
with pytest.raises(Exception, match="no MemWAL write spec"):
table.disable_memwal()
table.enable_memwal(MemWalWriteSpec.bucket("id", 8))
with pytest.raises(Exception, match="no RTE write spec"):
table.disable_rte()
table.enable_rte(RTEWriteSpec.bucket("id", 8))
def test_enable_rte_requires_primary_key(tmp_path):
_db, table = _make_table(tmp_path)
for spec in (
RTEWriteSpec.unsharded(),
RTEWriteSpec.bucket("id", 4),
RTEWriteSpec.identity("id"),
):
with pytest.raises(Exception, match="primary key"):
table.enable_rte(spec)
assert table.get_rte_spec() is None
def test_set_unsharded_spec(tmp_path):
@@ -79,12 +93,12 @@ def test_set_unsharded_spec(tmp_path):
# Lance MemWAL still requires a primary key on the dataset; Unsharded
# just skips per-row hashing.
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.unsharded())
table.disable_memwal()
table.enable_rte(RTEWriteSpec.unsharded())
table.disable_rte()
def test_memwal_spec_repr():
s = MemWalWriteSpec.bucket("id", 4)
def test_rte_spec_repr():
s = RTEWriteSpec.bucket("id", 4)
assert s.spec_type == "bucket"
assert s.column == "id"
assert s.num_buckets == 4
@@ -95,20 +109,20 @@ def test_memwal_spec_repr():
assert "id" in repr(s)
assert "4" in repr(s)
u = MemWalWriteSpec.unsharded()
u = RTEWriteSpec.unsharded()
assert u.spec_type == "unsharded"
assert u.column is None
assert u.num_buckets is None
assert "unsharded" in repr(u)
def test_memwal_spec_with_maintained_indexes():
s = MemWalWriteSpec.bucket("id", 4).with_maintained_indexes(["idx_a", "idx_b"])
def test_rte_spec_with_maintained_indexes():
s = RTEWriteSpec.bucket("id", 4).with_maintained_indexes(["idx_a", "idx_b"])
assert s.maintained_indexes == ["idx_a", "idx_b"]
@pytest.mark.asyncio
async def test_async_set_disable_memwal(tmp_path):
async def test_async_set_disable_rte(tmp_path):
db = await lancedb.connect_async(
tmp_path, read_consistency_interval=timedelta(seconds=0)
)
@@ -118,11 +132,11 @@ async def test_async_set_disable_memwal(tmp_path):
)
await table.set_unenforced_primary_key("id")
await table.enable_memwal(MemWalWriteSpec.bucket("id", 4))
await table.disable_memwal()
await table.enable_rte(RTEWriteSpec.bucket("id", 4))
await table.disable_rte()
# A second unset errors.
with pytest.raises(Exception, match="no MemWAL write spec"):
await table.disable_memwal()
with pytest.raises(Exception, match="no RTE write spec"):
await table.disable_rte()
def test_set_identity_spec(tmp_path):
@@ -130,12 +144,12 @@ def test_set_identity_spec(tmp_path):
# Identity sharding still requires an unenforced primary key on the
# table; it shards by the raw value of the given column.
table.set_unenforced_primary_key("id")
table.enable_memwal(MemWalWriteSpec.identity("v"))
table.disable_memwal()
table.enable_rte(RTEWriteSpec.identity("v"))
table.disable_rte()
def test_memwal_spec_identity_and_writer_config_defaults():
s = MemWalWriteSpec.identity("v")
def test_rte_spec_identity_and_writer_config_defaults():
s = RTEWriteSpec.identity("v")
assert s.spec_type == "identity"
assert s.column == "v"
assert s.num_buckets is None
@@ -146,12 +160,12 @@ def test_memwal_spec_identity_and_writer_config_defaults():
assert "durable_write" in repr(s)
def test_get_memwal_spec(tmp_path):
def test_get_rte_spec(tmp_path):
_db, table = _make_table(tmp_path)
table.set_unenforced_primary_key("id")
# None when nothing is installed.
assert table.get_memwal_spec() is None
assert table.get_rte_spec() is None
# A real scalar index is needed to name it as a maintained index.
table.create_index("id", config=BTree())
@@ -159,12 +173,12 @@ def test_get_memwal_spec(tmp_path):
# Bucket spec round-trips, including maintained indexes and writer config
# defaults.
table.enable_memwal(
MemWalWriteSpec.bucket("id", 4)
table.enable_rte(
RTEWriteSpec.bucket("id", 4)
.with_maintained_indexes([idx_name])
.with_writer_config_defaults({"durable_write": "false"})
)
spec = table.get_memwal_spec()
spec = table.get_rte_spec()
assert spec is not None
assert spec.spec_type == "bucket"
assert spec.column == "id"
@@ -173,30 +187,30 @@ def test_get_memwal_spec(tmp_path):
assert spec.writer_config_defaults == {"durable_write": "false"}
# After unset, None again.
table.disable_memwal()
assert table.get_memwal_spec() is None
table.disable_rte()
assert table.get_rte_spec() is None
# Identity round-trips (column recovered from the schema). An unnamed
# maintained set reads back as None: what is stored is the intent to
# maintain every index, not the set it resolves to now.
table.enable_memwal(MemWalWriteSpec.identity("id"))
spec = table.get_memwal_spec()
table.enable_rte(RTEWriteSpec.identity("id"))
spec = table.get_rte_spec()
assert spec.spec_type == "identity"
assert spec.column == "id"
assert spec.maintained_indexes is None
table.disable_memwal()
table.disable_rte()
# Unsharded round-trips (no routing column). Opting out is distinct from
# the unnamed default.
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([]))
spec = table.get_memwal_spec()
table.enable_rte(RTEWriteSpec.unsharded().with_maintained_indexes([]))
spec = table.get_rte_spec()
assert spec.spec_type == "unsharded"
assert spec.column is None
assert spec.maintained_indexes == []
@pytest.mark.asyncio
async def test_async_get_memwal_spec(tmp_path):
async def test_async_get_rte_spec(tmp_path):
db = await lancedb.connect_async(
tmp_path, read_consistency_interval=timedelta(seconds=0)
)
@@ -205,20 +219,21 @@ async def test_async_get_memwal_spec(tmp_path):
pa.RecordBatchReader.from_batches(SCHEMA, [_batch(["seed"], [0])]),
)
assert await table.get_memwal_spec() is None
assert await table.get_rte_spec() is None
await table.set_unenforced_primary_key("id")
# A real scalar index is needed to name it as a maintained index.
await table.create_index("id", config=BTree())
idx_name = (await table.list_indices())[0].name
await table.enable_memwal(
MemWalWriteSpec.bucket("id", 8).with_maintained_indexes([idx_name])
await table.enable_rte(
RTEWriteSpec.bucket("id", 8).with_maintained_indexes([idx_name])
)
spec = await table.get_memwal_spec()
spec = await table.get_rte_spec()
assert spec is not None
assert spec.spec_type == "bucket"
assert spec.column == "id"
assert spec.num_buckets == 8
assert spec.maintained_indexes == [idx_name]
await table.disable_memwal()
assert await table.get_memwal_spec() is None
await table.disable_rte()
assert await table.get_rte_spec() is None
+7 -7
View File
@@ -14,7 +14,7 @@ from lancedb.rerankers import RRFReranker
def _wal_table(schema: pa.Schema) -> mock.Mock:
"""A MemWAL-backed table a previous query has already been refused
"""An RTE-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
@@ -51,7 +51,7 @@ def test_wal_hybrid_projects_the_key_instead_of_asking_for_row_ids():
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.
# An RTE table rejects with_row_id before it plans.
assert leg._with_row_id is not True
@@ -81,15 +81,15 @@ def test_base_table_hybrid_still_joins_on_row_ids():
assert builder._vector_query._columns is None
def test_wal_hybrid_with_use_memwal_false_joins_on_row_ids():
builder = _wal_hybrid().use_memwal(False)
def test_wal_hybrid_with_use_rte_false_joins_on_row_ids():
builder = _wal_hybrid().use_rte(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_memwal"):
with pytest.raises(NotImplementedError, match="use_rte"):
builder._create_query_builders()
@@ -141,7 +141,7 @@ def test_wal_hybrid_refuses_row_id_even_with_return_score_all(row_id_first: bool
else:
builder = builder.rerank(reranker).with_row_id(True)
with pytest.raises(NotImplementedError, match="use_memwal"):
with pytest.raises(NotImplementedError, match="use_rte"):
builder._create_query_builders()
@@ -199,7 +199,7 @@ def test_fallback_explains_the_refusal_to_a_caller_who_asked_for_row_ids():
def run():
raise _REFUSAL
with pytest.raises(NotImplementedError, match="use_memwal") as info:
with pytest.raises(NotImplementedError, match="use_rte") as info:
with_pk_fallback(
run, mock.Mock(), fused=lambda: False, caller_requested_row_id=True
)
+3 -3
View File
@@ -73,10 +73,10 @@ class MockPermutationServer:
},
},
)
if path == f"/v1/table/{self.name}/get_memwal_spec/":
if path == f"/v1/table/{self.name}/get_rte_spec/":
self._read_body(request)
# Null spec: this table has no MemWAL write path.
return self._json(request, {"memwal_spec": None})
# Null spec: this table has no RTE write path.
return self._json(request, {"rte_spec": None})
if path == f"/v1/table/{self.name}/count_rows/":
self._read_body(request)
return self._json(request, self.num_rows)
+3 -3
View File
@@ -16,8 +16,8 @@ use query::{FTSQuery, HybridQuery, Query, VectorQuery};
use session::Session;
use table::{
AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken,
FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, MemWalWriteSpec, MergeResult,
PyBlobFile, RefreshColumnResult, RefreshMaterializedViewResult, Table,
FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, MergeResult, PyBlobFile,
RTEWriteSpec, RefreshColumnResult, RefreshMaterializedViewResult, Table,
UpdateFieldMetadataResult, UpdateResult,
};
@@ -104,7 +104,7 @@ pub fn _lancedb(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<UpdateFieldMetadataResult>()?;
m.add_class::<AddResult>()?;
m.add_class::<MergeResult>()?;
m.add_class::<MemWalWriteSpec>()?;
m.add_class::<RTEWriteSpec>()?;
m.add_class::<DeleteResult>()?;
m.add_class::<DropColumnsResult>()?;
m.add_class::<UpdateResult>()?;
+14 -14
View File
@@ -333,7 +333,7 @@ pub struct PyQueryRequest {
pub select_source_columns: Option<HashMap<String, String>>,
pub fast_search: Option<bool>,
pub with_row_id: Option<bool>,
pub use_memwal: Option<bool>,
pub use_rte: Option<bool>,
pub column: Option<String>,
pub query_vector: Option<PyQueryVectors>,
pub nprobes: Option<usize>,
@@ -366,7 +366,7 @@ impl From<AnyQuery> for PyQueryRequest {
select: PySelect(query_request.select),
fast_search: Some(query_request.fast_search),
with_row_id: Some(query_request.with_row_id),
use_memwal: query_request.use_memwal,
use_rte: query_request.use_rte,
column: None,
query_vector: None,
nprobes: None,
@@ -395,7 +395,7 @@ impl From<AnyQuery> for PyQueryRequest {
select: PySelect(vector_query.base.select),
fast_search: Some(vector_query.base.fast_search),
with_row_id: Some(vector_query.base.with_row_id),
use_memwal: vector_query.base.use_memwal,
use_rte: vector_query.base.use_rte,
column: vector_query.column,
query_vector: Some(PyQueryVectors(vector_query.query_vector)),
nprobes: vector_query.nprobes,
@@ -540,8 +540,8 @@ impl Query {
self.inner = self.inner.clone().fast_search();
}
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(enable);
pub fn use_rte(&mut self, enable: bool) {
self.inner = self.inner.clone().use_rte(enable);
}
pub fn with_row_id(&mut self) {
@@ -706,8 +706,8 @@ impl TakeQuery {
self.inner = self.inner.clone().with_row_id();
}
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(enable);
pub fn use_rte(&mut self, enable: bool) {
self.inner = self.inner.clone().use_rte(enable);
}
#[pyo3(signature = ())]
@@ -819,8 +819,8 @@ impl FTSQuery {
self.inner = self.inner.clone().fast_search();
}
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(enable);
pub fn use_rte(&mut self, enable: bool) {
self.inner = self.inner.clone().use_rte(enable);
}
pub fn with_row_id(&mut self) {
@@ -970,8 +970,8 @@ impl VectorQuery {
self.inner = self.inner.clone().fast_search();
}
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(enable);
pub fn use_rte(&mut self, enable: bool) {
self.inner = self.inner.clone().use_rte(enable);
}
pub fn with_row_id(&mut self) {
@@ -1178,9 +1178,9 @@ impl HybridQuery {
self.inner_fts.postfilter();
}
pub fn use_memwal(&mut self, enable: bool) {
self.inner_vec.use_memwal(enable);
self.inner_fts.use_memwal(enable);
pub fn use_rte(&mut self, enable: bool) {
self.inner_vec.use_rte(enable);
self.inner_fts.use_rte(enable);
}
pub fn add_query_vector(&mut self, vector: Bound<'_, PyAny>) -> PyResult<()> {
+59 -58
View File
@@ -41,13 +41,13 @@ use pyo3::{
mod scannable;
/// Convert `MemWalStats` to a Python dict, preserving the per-table-shard list.
/// Convert `RTEStats` to a Python dict, preserving the per-table-shard list.
///
/// Deliberately not flattened to a table-level summary: a table is N
/// table shards on one node, and the per-shard detail is the reason the
/// endpoint exists — flattening hides the single hot table shard someone opened
/// it to find.
fn memwal_stats_to_py(py: Python<'_>, stats: &lancedb::table::MemWalStats) -> PyResult<Py<PyDict>> {
fn rte_stats_to_py(py: Python<'_>, stats: &lancedb::table::RTEStats) -> PyResult<Py<PyDict>> {
let out = PyDict::new(py);
let table_shards = PyList::empty(py);
for b in &stats.table_shards {
@@ -393,8 +393,8 @@ fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
}
}
/// Specification selecting Lance's MemWAL LSM-style write path for
/// `merge_insert`.
/// Specification enabling the realtime engine (RTE), which routes
/// `merge_insert` through Lance's MemWAL LSM-style write path.
///
/// Constructed via the `bucket(...)`, `identity(...)`, or `unsharded()`
/// classmethods, then optionally chain `with_maintained_indexes(...)` and
@@ -402,17 +402,17 @@ fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
/// table has, including ones created later.
#[pyclass(module = "lancedb._lancedb", from_py_object)]
#[derive(Clone, Debug)]
pub struct MemWalWriteSpec {
inner: lancedb::table::MemWalWriteSpec,
pub struct RTEWriteSpec {
inner: lancedb::table::RTEWriteSpec,
}
#[pymethods]
impl MemWalWriteSpec {
impl RTEWriteSpec {
/// Hash-bucket sharding by the unenforced primary key column.
#[staticmethod]
pub fn bucket(column: String, num_buckets: u32) -> Self {
Self {
inner: lancedb::table::MemWalWriteSpec::bucket(column, num_buckets),
inner: lancedb::table::RTEWriteSpec::bucket(column, num_buckets),
}
}
@@ -426,20 +426,20 @@ impl MemWalWriteSpec {
#[staticmethod]
pub fn identity(column: String) -> Self {
Self {
inner: lancedb::table::MemWalWriteSpec::identity(column),
inner: lancedb::table::RTEWriteSpec::identity(column),
}
}
/// No sharding — every `merge_insert` call writes to a single
/// MemWAL shard.
/// RTE shard.
#[staticmethod]
pub fn unsharded() -> Self {
Self {
inner: lancedb::table::MemWalWriteSpec::unsharded(),
inner: lancedb::table::RTEWriteSpec::unsharded(),
}
}
/// Set which indexes the MemWAL maintains. `None` (the default) is
/// Set which indexes the RTE 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))]
@@ -459,33 +459,33 @@ impl MemWalWriteSpec {
pub fn __repr__(&self) -> String {
match &self.inner {
lancedb::table::MemWalWriteSpec::Bucket {
lancedb::table::RTEWriteSpec::Bucket {
column,
num_buckets,
maintained_indexes,
writer_config_defaults,
} => format!(
"MemWalWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
"RTEWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
column,
num_buckets,
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
lancedb::table::MemWalWriteSpec::Identity {
lancedb::table::RTEWriteSpec::Identity {
column,
maintained_indexes,
writer_config_defaults,
} => format!(
"MemWalWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
"RTEWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
column,
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
lancedb::table::MemWalWriteSpec::Unsharded {
lancedb::table::RTEWriteSpec::Unsharded {
maintained_indexes,
writer_config_defaults,
} => format!(
"MemWalWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
"RTEWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
@@ -497,9 +497,9 @@ impl MemWalWriteSpec {
#[getter]
pub fn spec_type(&self) -> &'static str {
match &self.inner {
lancedb::table::MemWalWriteSpec::Bucket { .. } => "bucket",
lancedb::table::MemWalWriteSpec::Identity { .. } => "identity",
lancedb::table::MemWalWriteSpec::Unsharded { .. } => "unsharded",
lancedb::table::RTEWriteSpec::Bucket { .. } => "bucket",
lancedb::table::RTEWriteSpec::Identity { .. } => "identity",
lancedb::table::RTEWriteSpec::Unsharded { .. } => "unsharded",
}
}
@@ -507,9 +507,9 @@ impl MemWalWriteSpec {
#[getter]
pub fn column(&self) -> Option<String> {
match &self.inner {
lancedb::table::MemWalWriteSpec::Bucket { column, .. }
| lancedb::table::MemWalWriteSpec::Identity { column, .. } => Some(column.clone()),
lancedb::table::MemWalWriteSpec::Unsharded { .. } => None,
lancedb::table::RTEWriteSpec::Bucket { column, .. }
| lancedb::table::RTEWriteSpec::Identity { column, .. } => Some(column.clone()),
lancedb::table::RTEWriteSpec::Unsharded { .. } => None,
}
}
@@ -517,12 +517,12 @@ impl MemWalWriteSpec {
#[getter]
pub fn num_buckets(&self) -> Option<u32> {
match &self.inner {
lancedb::table::MemWalWriteSpec::Bucket { num_buckets, .. } => Some(*num_buckets),
lancedb::table::RTEWriteSpec::Bucket { num_buckets, .. } => Some(*num_buckets),
_ => None,
}
}
/// Indexes the MemWAL keeps up to date, or `None` for every supported one.
/// Indexes the RTE keeps up to date, or `None` for every supported one.
#[getter]
pub fn maintained_indexes(&self) -> Option<Vec<String>> {
self.inner.maintained_indexes().map(<[String]>::to_vec)
@@ -535,14 +535,14 @@ impl MemWalWriteSpec {
}
}
impl From<MemWalWriteSpec> for lancedb::table::MemWalWriteSpec {
fn from(spec: MemWalWriteSpec) -> Self {
impl From<RTEWriteSpec> for lancedb::table::RTEWriteSpec {
fn from(spec: RTEWriteSpec) -> Self {
spec.inner
}
}
impl From<lancedb::table::MemWalWriteSpec> for MemWalWriteSpec {
fn from(inner: lancedb::table::MemWalWriteSpec) -> Self {
impl From<lancedb::table::RTEWriteSpec> for RTEWriteSpec {
fn from(inner: lancedb::table::RTEWriteSpec) -> Self {
Self { inner }
}
}
@@ -1770,8 +1770,8 @@ impl Table {
if let Some(use_index) = parameters.use_index {
builder.use_index(use_index);
}
if let Some(use_memwal) = parameters.use_memwal {
builder.use_memwal(use_memwal);
if let Some(use_rte) = parameters.use_rte {
builder.use_rte(use_rte);
}
if let Some(validate_single_shard) = parameters.validate_single_shard {
builder.validate_single_shard(validate_single_shard);
@@ -1796,29 +1796,29 @@ impl Table {
})
}
pub fn enable_memwal<'a>(
pub fn enable_rte<'a>(
self_: PyRef<'a, Self>,
spec: MemWalWriteSpec,
spec: RTEWriteSpec,
) -> PyResult<Bound<'a, PyAny>> {
let inner = self_.inner_ref()?.clone();
let native_spec = lancedb::table::MemWalWriteSpec::from(spec);
let native_spec = lancedb::table::RTEWriteSpec::from(spec);
future_into_py(self_.py(), async move {
inner.enable_memwal(native_spec).await.infer_error()
inner.enable_rte(native_spec).await.infer_error()
})
}
pub fn disable_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn disable_rte(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.disable_memwal().await.infer_error()
inner.disable_rte().await.infer_error()
})
}
pub fn get_memwal_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn get_rte_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let spec = inner.get_memwal_spec().await.infer_error()?;
Ok(spec.map(MemWalWriteSpec::from))
let spec = inner.get_rte_spec().await.infer_error()?;
Ok(spec.map(RTEWriteSpec::from))
})
}
@@ -1837,54 +1837,55 @@ impl Table {
Ok(())
}
/// Converge the table's MemWAL write path into its base table.
/// Converge the table's RTE write path into its base table.
///
/// Best-effort: with writes flowing, new rows may land after the last
/// pass. Errors if the table stops making progress.
pub fn checkpoint_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn checkpoint_rte(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.checkpoint_memwal().await.infer_error()
inner.checkpoint_rte().await.infer_error()
})
}
/// Freeze every table shard's active memtable into an SSTable.
pub fn flush_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn flush_rte(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.flush_memwal().await.infer_error()
})
future_into_py(
self_.py(),
async move { inner.flush_rte().await.infer_error() },
)
}
/// Trigger a background SSTable compaction pass per table shard. Returns once the
/// passes are dispatched, not once they finish — watch `get_memwal_stats`.
pub fn compact_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
/// passes are dispatched, not once they finish — watch `get_rte_stats`.
pub fn compact_rte(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.compact_memwal().await.infer_error()
inner.compact_rte().await.infer_error()
})
}
/// Live MemWAL state, or `None` when the MemWAL write path is not enabled.
/// Live RTE state, or `None` when the RTE write path is not enabled.
#[pyo3(signature = (include_sstable_rows=false))]
pub fn get_memwal_stats(
pub fn get_rte_stats(
self_: PyRef<'_, Self>,
include_sstable_rows: bool,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let stats = inner
.get_memwal_stats(include_sstable_rows)
.get_rte_stats(include_sstable_rows)
.await
.infer_error()?;
Python::attach(|py| stats.map(|s| memwal_stats_to_py(py, &s)).transpose())
Python::attach(|py| stats.map(|s| rte_stats_to_py(py, &s)).transpose())
})
}
pub fn close_memwal_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn close_rte_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.close_memwal_writers().await.infer_error()
inner.close_rte_writers().await.infer_error()
})
}
@@ -2216,7 +2217,7 @@ pub struct MergeInsertParams {
when_not_matched_by_source_condition_expr: Option<PyExpr>,
timeout: Option<std::time::Duration>,
use_index: Option<bool>,
use_memwal: Option<bool>,
use_rte: Option<bool>,
validate_single_shard: Option<bool>,
}