rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-17 01:02:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge main into repair branch and preserve beta.19 validity fix

Browse Source
This commit is contained in:
Gatefixer
2026-10-16 21:49:42 +00:00
parent d46525e056 91541408e7
commit b604b425bb
83 changed files with 3021 additions and 2019 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/python/lancedb/__init__.py
+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 LsmWriteSpec
from ._lancedb import MemWalWriteSpec
from ._lancedb import tokenize as _tokenize
from .common import URI, sanitize_uri
from urllib.parse import urlparse
@@ -610,7 +610,7 @@ __all__ = [
"SqlQuery",
"LanceDBConnection",
"LanceNamespaceDBConnection",
"LsmWriteSpec",
"MemWalWriteSpec",
"RemoteDBConnection",
"Session",
"Table",
python/python/lancedb/_lancedb.pyi
+24 -20
View File
@@ -538,17 +538,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 set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
async def unset_lsm_write_spec(self) -> None: ...
async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
async def enable_memwal(self, spec: MemWalWriteSpec) -> None: ...
async def disable_memwal(self) -> None: ...
async def get_memwal_spec(self) -> Optional[MemWalWriteSpec]: ...
# WAL-PK-FUSION: delete both.
def hybrid_pk_fusion_learned(self) -> bool: ...
def note_hybrid_pk_fusion(self) -> None: ...
async def checkpoint_lsm(self) -> None: ...
async def flush_lsm(self) -> None: ...
async def compact_lsm(self) -> None: ...
async def get_lsm_stats(self, include_generation_rows: bool) -> Optional[dict]: ...
async def close_lsm_writers(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: ...
@property
def tags(self) -> Tags: ...
@property
@@ -661,7 +661,7 @@ class Query:
def fast_search(self): ...
def with_row_id(self): ...
def postfilter(self): ...
def use_lsm(self, enable: bool): ...
def use_memwal(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]]): ...
@@ -678,7 +678,7 @@ class Query:
class TakeQuery:
def select(self, columns: List[str]): ...
def with_row_id(self): ...
def use_lsm(self, enable: bool): ...
def use_memwal(self, enable: bool): ...
async def output_schema(self) -> pa.Schema: ...
async def execute(self) -> RecordBatchStream: ...
async def explain_plan(self, verbose: Optional[bool]) -> str: ...
@@ -697,7 +697,7 @@ class FTSQuery:
def fast_search(self): ...
def with_row_id(self): ...
def postfilter(self): ...
def use_lsm(self, enable: bool): ...
def use_memwal(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: ...
@@ -725,7 +725,7 @@ class VectorQuery:
def column(self, column: str): ...
def distance_type(self, distance_type: str): ...
def postfilter(self): ...
def use_lsm(self, enable: bool): ...
def use_memwal(self, enable: bool): ...
def refine_factor(self, refine_factor: int): ...
def nprobes(self, nprobes: int): ...
def search_effort(self, search_effort: Optional[float]): ...
@@ -750,7 +750,7 @@ class HybridQuery:
def fast_search(self): ...
def with_row_id(self): ...
def postfilter(self): ...
def use_lsm(self, enable: bool): ...
def use_memwal(self, enable: bool): ...
def distance_type(self, distance_type: str): ...
def refine_factor(self, refine_factor: int): ...
def nprobes(self, nprobes: int): ...
@@ -778,7 +778,7 @@ class PyQueryRequest:
select_source_columns: Optional[Dict[str, str]]
fast_search: Optional[bool]
with_row_id: Optional[bool]
use_lsm: Optional[bool]
use_memwal: Optional[bool]
column: Optional[str]
query_vector: Optional[List[pa.Array]]
nprobes: Optional[int]
@@ -839,22 +839,26 @@ class FtsToken:
text: str
position: int
class LsmWriteSpec:
class MemWalWriteSpec:
"""Specification selecting Lance's MemWAL LSM-style write path for
`merge_insert`."""
@staticmethod
def bucket(column: str, num_buckets: int) -> "LsmWriteSpec": ...
def bucket(column: str, num_buckets: int) -> "MemWalWriteSpec": ...
@staticmethod
def identity(column: str) -> "LsmWriteSpec": ...
def identity(column: str) -> "MemWalWriteSpec": ...
@staticmethod
def unsharded() -> "LsmWriteSpec": ...
def with_maintained_indexes(self, indexes: Optional[List[str]]) -> "LsmWriteSpec":
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
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]) -> "LsmWriteSpec":
def with_writer_config_defaults(
self, defaults: Dict[str, str]
) -> "MemWalWriteSpec":
"""Return a copy of this spec recording the given default
`ShardWriter` configuration in the MemWAL index."""
...
python/python/lancedb/_wal_hybrid.py
+4 -4
View File
@@ -76,7 +76,7 @@ _WAL_ROW_ID_REFUSAL = "does not support with_row_id"
WAL_ROW_ID_UNSUPPORTED = (
"hybrid search on a MemWAL table cannot return _rowid: the fresh tier has "
"no stable row id, and the ids the fusion joins on are synthesized from "
"the primary key. Set use_lsm(False) to read the base table only (results "
"the primary key. Set use_memwal(False) to read the base table only (results "
"will exclude un-compacted MemWAL data)"
)
@@ -151,13 +151,13 @@ def _stamp_surrogate_row_ids(
return stamped
def pk_fusion_learned(table, use_lsm: Optional[bool]) -> bool:
def pk_fusion_learned(table, use_memwal: Optional[bool]) -> bool:
"""Whether a previous hybrid query on `table` was refused `_rowid`, so this
one should go straight to the primary key.
Learned from a refusal, never probed, so asking is free.
"""
if table is None or use_lsm is False:
if table is None or use_memwal is False:
return False
# Only an unambiguous True forks to the key: a table type that answers the
# question loosely keeps the `_rowid` path.
@@ -192,7 +192,7 @@ class PkFusion:
raise NotImplementedError(
"hybrid search cannot project a blob column on a MemWAL table: "
"fetching blobs needs a real _rowid, and the fresh tier has no "
"stable one. Set use_lsm(False) to read the base table only "
"stable one. Set use_memwal(False) to read the base table only "
"(results will exclude un-compacted MemWAL data)"
)
return cls(_pk_columns(schema))
python/python/lancedb/merge.py
+7 -7
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_lsm = None
self._use_memwal = 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_lsm(self, enable: bool) -> LanceMergeInsertBuilder:
def use_memwal(self, enable: bool) -> LanceMergeInsertBuilder:
"""
Control MemWAL routing for this merge.
By default (unset), a `merge_insert` on a table with an LSM write spec is
By default (unset), a `merge_insert` on a table with a MemWAL 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 LSM
``True`` forces MemWAL routing and errors if the table has no MemWAL
write spec. ``False`` forces the standard write path even when a spec
is set.
"""
self._use_lsm = enable
self._use_memwal = 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 an LSM write spec, every row in a `merge_insert` call
When a table has a MemWAL 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 an LSM write spec.
Has no effect on tables without a MemWAL write spec.
Parameters
----------
python/python/lancedb/permutation.py
+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 an LSM write spec are rejected: unflushed rows have no row id.
Tables with a MemWAL write spec are rejected: unflushed rows have no row id.
"""
self._async = async_permutation_builder(table)
python/python/lancedb/query.py
+27 -27
View File
@@ -890,7 +890,7 @@ class Query(pydantic.BaseModel):
# MemWAL 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_lsm: Optional[bool] = None
use_memwal: 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_lsm is a genuine tri-state (None / True / False); preserve it as-is
# use_memwal is a genuine tri-state (None / True / False); preserve it as-is
# so a round-tripped query keeps an explicit False.
query.use_lsm = req.use_lsm
query.use_memwal = req.use_memwal
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_lsm = None
self._use_memwal = None
self._vector = None
self._text = None
self._ef = None
@@ -1520,19 +1520,19 @@ class LanceQueryBuilder(ABC):
self._fragment_ids = fragment_ids
return self
def use_lsm(self, enable: bool) -> Self:
def use_memwal(self, enable: bool) -> Self:
"""Control MemWAL LSM read routing for this query.
By default (unset), a query against a table with an LSM write spec is
By default (unset), a query against a table with a MemWAL 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 + flushed generations); a table without a
table (active/frozen memtables + SSTables); a table without a
spec reads the base table.
Parameters
----------
enable : bool
``True`` forces the LSM scanner and errors if the table has no LSM
``True`` forces the LSM scanner and errors if the table has no MemWAL
write spec. ``False`` bypasses the MemWAL and reads the base table
only, even when a spec is present.
@@ -1541,7 +1541,7 @@ class LanceQueryBuilder(ABC):
LanceQueryBuilder
The LanceQueryBuilder object.
"""
self._use_lsm = enable
self._use_memwal = enable
return self
def explain_plan(self, verbose: Optional[bool] = False) -> str:
@@ -2055,7 +2055,7 @@ class LanceVectorQueryBuilder(LanceQueryBuilder):
with_row_address=self._with_row_address,
fragments=self._fragments,
fragment_ids=self._fragment_ids,
use_lsm=self._use_lsm,
use_memwal=self._use_memwal,
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_lsm=self._use_lsm,
use_memwal=self._use_memwal,
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_lsm=self._use_lsm,
use_memwal=self._use_memwal,
offset=self._offset,
order_by=self._order_by,
)
@@ -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_lsm):
if _wal_hybrid.pk_fusion_learned(self._table, self._use_memwal):
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_lsm is not None:
self._vector_query.use_lsm(self._use_lsm)
self._fts_query.use_lsm(self._use_lsm)
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._phrase_query:
self._fts_query.phrase_query(True)
if self._distance_type:
@@ -3723,25 +3723,25 @@ class AsyncStandardQuery(AsyncQueryBase):
self._inner.fast_search()
return self
def use_lsm(self, enable: bool) -> Self:
def use_memwal(self, enable: bool) -> Self:
"""
Control MemWAL LSM read routing for this query.
By default (unset), a query against a table with an LSM write spec (see
[AsyncTable.set_lsm_write_spec][lancedb.table.AsyncTable.set_lsm_write_spec])
By default (unset), a query against a table with a MemWAL write spec (see
[AsyncTable.enable_memwal][lancedb.table.AsyncTable.enable_memwal])
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 flushed generations),
table (the active/frozen in-memory memtables and the SSTables),
deduplicated by primary key; a table without a spec reads the base table.
Parameters
----------
enable : bool
``True`` forces the LSM scanner and errors if the table has no LSM
``True`` forces the LSM scanner and errors if the table has no MemWAL
write spec. ``False`` bypasses the MemWAL and reads the base table
only, even when a spec is present.
"""
self._inner.use_lsm(enable)
self._inner.use_memwal(enable)
return self
def postfilter(self) -> Self:
@@ -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_lsm):
if _wal_hybrid.pk_fusion_learned(self._table, fts_req.use_memwal):
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_lsm(self, enable: bool) -> "AsyncTakeQuery":
def use_memwal(self, enable: bool) -> "AsyncTakeQuery":
"""Control MemWAL LSM read routing for this take query.
``False`` bypasses the MemWAL 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_lsm(enable)
self._inner.use_memwal(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_lsm(self, enable: bool) -> Self:
def use_memwal(self, enable: bool) -> Self:
"""
Control MemWAL LSM read routing for this query.
``False`` bypasses the MemWAL 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_lsm(enable)
self._inner.use_memwal(enable)
return self
def with_row_address(self, with_row_address: bool = True) -> Self:
python/python/lancedb/remote/table.py
+25 -25
View File
@@ -31,7 +31,7 @@ from lancedb._lancedb import (
DropColumnsResult,
FtsToken,
IndexConfig,
LsmWriteSpec,
MemWalWriteSpec,
MergeResult,
UpdateResult,
)
@@ -1034,20 +1034,20 @@ class RemoteTable(Table):
return LOOP.run(self._table.drop_columns(columns))
def set_unenforced_primary_key(self, columns: Union[str, Iterable[str]]) -> None:
"""Not supported for remote tables."""
"""Set the unenforced primary key for this table to a single column."""
return LOOP.run(self._table.set_unenforced_primary_key(columns))
def set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:
"""Install an LsmWriteSpec."""
return LOOP.run(self._table.set_lsm_write_spec(spec))
def enable_memwal(self, spec: "MemWalWriteSpec") -> None:
"""Install an MemWalWriteSpec."""
return LOOP.run(self._table.enable_memwal(spec))
def unset_lsm_write_spec(self) -> None:
"""Remove the LsmWriteSpec."""
return LOOP.run(self._table.unset_lsm_write_spec())
def disable_memwal(self) -> None:
"""Remove the MemWalWriteSpec."""
return LOOP.run(self._table.disable_memwal())
def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
"""Read the installed LsmWriteSpec, or ``None``."""
return LOOP.run(self._table.get_lsm_write_spec())
def get_memwal_spec(self) -> Optional["MemWalWriteSpec"]:
"""Read the installed MemWalWriteSpec, or ``None``."""
return LOOP.run(self._table.get_memwal_spec())
# WAL-PK-FUSION: delete both hooks.
def _hybrid_pk_fusion_learned(self) -> bool:
@@ -1057,31 +1057,31 @@ class RemoteTable(Table):
def _note_hybrid_pk_fusion(self) -> None:
self._table._note_hybrid_pk_fusion()
def checkpoint_lsm(self) -> None:
def checkpoint_memwal(self) -> None:
"""Synchronous version of
[`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
return LOOP.run(self._table.checkpoint_lsm())
[`AsyncTable.checkpoint_memwal`][lancedb.AsyncTable.checkpoint_memwal]."""
return LOOP.run(self._table.checkpoint_memwal())
def flush_lsm(self) -> None:
def flush_memwal(self) -> None:
"""Synchronous version of
[`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
return LOOP.run(self._table.flush_lsm())
[`AsyncTable.flush_memwal`][lancedb.AsyncTable.flush_memwal]."""
return LOOP.run(self._table.flush_memwal())
def compact_lsm(self) -> None:
def compact_memwal(self) -> None:
"""Synchronous version of
[`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
return LOOP.run(self._table.compact_lsm())
[`AsyncTable.compact_memwal`][lancedb.AsyncTable.compact_memwal]."""
return LOOP.run(self._table.compact_memwal())
def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
def get_memwal_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
"""Synchronous version of
[`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
[`AsyncTable.get_memwal_stats`][lancedb.AsyncTable.get_memwal_stats]."""
return LOOP.run(
self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
self._table.get_memwal_stats(include_sstable_rows=include_sstable_rows)
)
def close_lsm_writers(self) -> None:
def close_memwal_writers(self) -> None:
"""No-op for remote tables (no local shard writers)."""
return LOOP.run(self._table.close_lsm_writers())
return LOOP.run(self._table.close_memwal_writers())
def drop_index(self, index_name: str):
return LOOP.run(self._table.drop_index(index_name))
python/python/lancedb/table.py
+84 -83
View File
@@ -223,7 +223,7 @@ if TYPE_CHECKING:
DeleteResult,
DropColumnsResult,
FtsToken,
LsmWriteSpec,
MemWalWriteSpec,
MergeResult,
UpdateResult,
)
@@ -2190,8 +2190,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
@@ -4827,47 +4828,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 set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:
"""Install an LsmWriteSpec. See
[`AsyncTable.set_lsm_write_spec`][lancedb.AsyncTable.set_lsm_write_spec]."""
return LOOP.run(self._table.set_lsm_write_spec(spec))
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 unset_lsm_write_spec(self) -> None:
"""Remove the LsmWriteSpec. See
[`AsyncTable.unset_lsm_write_spec`][lancedb.AsyncTable.unset_lsm_write_spec]."""
return LOOP.run(self._table.unset_lsm_write_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 get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
"""Read the installed LsmWriteSpec, or ``None``. See
[`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
return LOOP.run(self._table.get_lsm_write_spec())
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 checkpoint_lsm(self) -> None:
def checkpoint_memwal(self) -> None:
"""Synchronous version of
[`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
return LOOP.run(self._table.checkpoint_lsm())
[`AsyncTable.checkpoint_memwal`][lancedb.AsyncTable.checkpoint_memwal]."""
return LOOP.run(self._table.checkpoint_memwal())
def flush_lsm(self) -> None:
def flush_memwal(self) -> None:
"""Synchronous version of
[`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
return LOOP.run(self._table.flush_lsm())
[`AsyncTable.flush_memwal`][lancedb.AsyncTable.flush_memwal]."""
return LOOP.run(self._table.flush_memwal())
def compact_lsm(self) -> None:
def compact_memwal(self) -> None:
"""Synchronous version of
[`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
return LOOP.run(self._table.compact_lsm())
[`AsyncTable.compact_memwal`][lancedb.AsyncTable.compact_memwal]."""
return LOOP.run(self._table.compact_memwal())
def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
def get_memwal_stats(self, *, include_sstable_rows: bool = False) -> Optional[dict]:
"""Synchronous version of
[`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
[`AsyncTable.get_memwal_stats`][lancedb.AsyncTable.get_memwal_stats]."""
return LOOP.run(
self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
self._table.get_memwal_stats(include_sstable_rows=include_sstable_rows)
)
def close_lsm_writers(self) -> None:
def close_memwal_writers(self) -> None:
"""Close cached MemWAL shard writers. See
[`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
return LOOP.run(self._table.close_lsm_writers())
[`AsyncTable.close_memwal_writers`][lancedb.AsyncTable.close_memwal_writers]."""
return LOOP.run(self._table.close_memwal_writers())
def uses_v2_manifest_paths(self) -> bool:
"""
@@ -5525,18 +5526,18 @@ class AsyncTable:
columns = list(columns)
await self._inner.set_unenforced_primary_key(columns)
async def set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:
"""Install an LsmWriteSpec on this table.
async def enable_memwal(self, spec: "MemWalWriteSpec") -> None:
"""Install an MemWalWriteSpec on this table.
The spec selects Lance's MemWAL LSM-style write path for future
`merge_insert` calls. ``LsmWriteSpec`` chooses one of three sharding
`merge_insert` calls. ``MemWalWriteSpec`` chooses one of three sharding
strategies:
- ``LsmWriteSpec.bucket(column, num_buckets)`` — hash-bucket writes by
- ``MemWalWriteSpec.bucket(column, num_buckets)`` — hash-bucket writes by
the single-column unenforced primary key.
- ``LsmWriteSpec.identity(column)`` — shard by the raw value of a
- ``MemWalWriteSpec.identity(column)`` — shard by the raw value of a
scalar column.
- ``LsmWriteSpec.unsharded()`` — route every write to a single shard.
- ``MemWalWriteSpec.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
@@ -5551,30 +5552,30 @@ class AsyncTable:
Parameters
----------
spec : LsmWriteSpec
spec : MemWalWriteSpec
The sharding spec to install.
Examples
--------
>>> from lancedb import LsmWriteSpec
>>> from lancedb import MemWalWriteSpec
>>> # table.set_unenforced_primary_key("id")
>>> # table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 16))
>>> # table.enable_memwal(MemWalWriteSpec.bucket("id", 16))
"""
await self._inner.set_lsm_write_spec(spec)
await self._inner.enable_memwal(spec)
async def unset_lsm_write_spec(self) -> None:
"""Remove the LsmWriteSpec from this table.
async def disable_memwal(self) -> None:
"""Remove the MemWalWriteSpec from this table.
Reverts to the standard `merge_insert` write path. Errors if no spec
is currently set.
"""
await self._inner.unset_lsm_write_spec()
await self._inner.disable_memwal()
async def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
"""Read the LsmWriteSpec currently installed on this table.
async def get_memwal_spec(self) -> Optional["MemWalWriteSpec"]:
"""Read the MemWalWriteSpec currently installed on this table.
Returns ``None`` when the LSM write path is not enabled at all — no
spec has been set, or one was removed with `unset_lsm_write_spec`.
Returns ``None`` when the MemWAL write path is not enabled at all — no
spec has been set, or one was removed with `disable_memwal`.
That is a different answer from a spec whose ``maintained_indexes``
is ``None``, which is an installed spec selecting indexes
automatically.
@@ -5584,7 +5585,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_lsm_write_spec()
return await self._inner.get_memwal_spec()
# WAL-PK-FUSION: delete both hooks.
def _hybrid_pk_fusion_learned(self) -> bool:
@@ -5595,19 +5596,19 @@ class AsyncTable:
"""See [`Table._note_hybrid_pk_fusion`][lancedb.table.Table]."""
self._inner.note_hybrid_pk_fusion()
async def checkpoint_lsm(self) -> None:
"""Converge this table's LSM write path into its base table.
async def checkpoint_memwal(self) -> None:
"""Converge this table's MemWAL write path into its base table.
One flush, sealing every memtable into L0, then compaction triggers
until every generation that existed at that moment has reached base.
The loop runs client-side, reading progress from ``get_lsm_stats``.
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``.
Best-effort: generations created *while* it runs are deliberately not
Best-effort: SSTables created *while* it runs are deliberately not
waited on, which is what lets it terminate on a table taking writes.
Idempotent and safe on a cadence.
There is no deadline, and the caller owns that. It returns when the
target generations are gone, raises on a terminal server fault, and
target SSTables are gone, raises on a terminal server fault, and
otherwise waits however long the server takes. A slow table and a
stuck one are the same picture from the client: the compactor pool is
shared across every table on the node, so a checkpoint queued behind
@@ -5615,56 +5616,56 @@ class AsyncTable:
``asyncio.wait_for`` for a wall-clock bound; abandoning it partway
costs nothing.
"""
await self._inner.checkpoint_lsm()
await self._inner.checkpoint_memwal()
async def flush_lsm(self) -> None:
"""Seal every bucket's active memtable into L0.
async def flush_memwal(self) -> None:
"""Freeze every table shard's active memtable into an SSTable.
Does not touch the base table — moving L0 into base is
`compact_lsm`. On a node that has not claimed this table, this claims
Does not touch the base table — compacting SSTables into base is
`compact_memwal`. On a node that has not claimed this table, this claims
it and replays its WAL log first.
"""
await self._inner.flush_lsm()
await self._inner.flush_memwal()
async def compact_lsm(self) -> None:
"""Trigger a background L0 to base compaction pass per bucket.
async def compact_memwal(self) -> None:
"""Trigger a background SSTable compaction pass per table shard.
Returns once the passes are dispatched, not once they finish: watch
``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop
until the current L0 has reached base.
``get_memwal_stats`` for progress, or use ``checkpoint_memwal`` to loop
until the current SSTables have reached base.
"""
await self._inner.compact_lsm()
await self._inner.compact_memwal()
async def get_lsm_stats(
self, *, include_generation_rows: bool = False
async def get_memwal_stats(
self, *, include_sstable_rows: bool = False
) -> Optional[dict]:
"""Read live per-bucket LSM state.
"""Read live per-table-shard MemWAL state.
Answers "how far behind is my fresh tier", "which bucket is hot", and
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 LSM write path is not enabled.
Returns ``None`` only when the MemWAL write path is not enabled.
Parameters
----------
include_generation_rows
Report a row count per L0 generation. Off by default: each count
opens an uncached Lance dataset, and ``checkpoint_lsm`` polls this
include_sstable_rows
Report a row count per SSTable. Off by default: each count
opens an uncached Lance dataset, and ``checkpoint_memwal`` polls this
needing only generation numbers.
"""
return await self._inner.get_lsm_stats(include_generation_rows)
return await self._inner.get_memwal_stats(include_sstable_rows)
async def close_lsm_writers(self) -> None:
async def close_memwal_writers(self) -> None:
"""Drain and close any cached MemWAL shard writers for this table.
When an LSM write spec is installed, `merge_insert` opens MemWAL shard
When a MemWAL write spec is installed, `merge_insert` opens MemWAL 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_lsm_writers()
await self._inner.close_memwal_writers()
@property
def name(self) -> str:
@@ -6553,8 +6554,8 @@ class AsyncTable:
take_query = self.take_offsets(query.take_offsets)
if query.columns:
take_query = take_query.select(query.columns)
if query.use_lsm is not None:
take_query = take_query.use_lsm(query.use_lsm)
if query.use_memwal is not None:
take_query = take_query.use_memwal(query.use_memwal)
if query.with_row_id:
take_query = take_query.with_row_id()
return take_query
@@ -6570,8 +6571,8 @@ class AsyncTable:
async_query = async_query.where(query.filter)
if query.fast_search:
async_query = async_query.fast_search()
if query.use_lsm is not None:
async_query = async_query.use_lsm(query.use_lsm)
if query.use_memwal is not None:
async_query = async_query.use_memwal(query.use_memwal)
if query.with_row_id:
async_query = async_query.with_row_id()
if query.order_by:
@@ -6697,7 +6698,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_lsm=merge._use_lsm,
use_memwal=merge._use_memwal,
validate_single_shard=merge._validate_single_shard,
),
)
python/python/tests/test_lsm_write_spec.py → python/python/tests/test_memwal_spec.py
+69 -54
View File
@@ -1,8 +1,8 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Tests for installing and clearing an LsmWriteSpec via
`Table.set_lsm_write_spec` / `Table.unset_lsm_write_spec`.
"""Tests for installing and clearing an MemWalWriteSpec via
`Table.enable_memwal` / `Table.disable_memwal`.
"""
from datetime import timedelta
@@ -10,7 +10,7 @@ from datetime import timedelta
import lancedb
import pyarrow as pa
import pytest
from lancedb._lancedb import LsmWriteSpec
from lancedb._lancedb import MemWalWriteSpec
from lancedb.index import BTree
SCHEMA = pa.schema(
@@ -21,9 +21,9 @@ SCHEMA = pa.schema(
)
def test_lsm_write_spec_module_metadata():
assert lancedb.LsmWriteSpec is LsmWriteSpec
assert LsmWriteSpec.__module__ == "lancedb._lancedb"
def test_memwal_spec_module_metadata():
assert lancedb.MemWalWriteSpec is MemWalWriteSpec
assert MemWalWriteSpec.__module__ == "lancedb._lancedb"
def _batch(ids, vs):
@@ -43,35 +43,49 @@ def _make_table(tmp_path):
return db, table
def test_set_lsm_write_spec_validates(tmp_path):
def test_enable_memwal_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.set_lsm_write_spec(LsmWriteSpec.bucket("id", 0))
table.enable_memwal(MemWalWriteSpec.bucket("id", 0))
with pytest.raises(Exception, match="num_buckets"):
table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 1025))
table.enable_memwal(MemWalWriteSpec.bucket("id", 1025))
# Happy path, then a set over the installed spec is refused.
table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 4))
table.enable_memwal(MemWalWriteSpec.bucket("id", 4))
with pytest.raises(Exception, match="already set"):
table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 8))
table.enable_memwal(MemWalWriteSpec.bucket("id", 8))
def test_unset_lsm_write_spec(tmp_path):
def test_disable_memwal(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 LSM write spec"):
table.unset_lsm_write_spec()
with pytest.raises(Exception, match="no MemWAL write spec"):
table.disable_memwal()
# Install a spec, then remove it; afterwards a fresh spec can be set.
table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 4))
table.unset_lsm_write_spec()
table.enable_memwal(MemWalWriteSpec.bucket("id", 4))
table.disable_memwal()
# A second unset errors — there is no spec left to remove.
with pytest.raises(Exception, match="no LSM write spec"):
table.unset_lsm_write_spec()
table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 8))
with pytest.raises(Exception, match="no MemWAL write spec"):
table.disable_memwal()
table.enable_memwal(MemWalWriteSpec.bucket("id", 8))
def test_enable_memwal_requires_primary_key(tmp_path):
_db, table = _make_table(tmp_path)
for spec in (
MemWalWriteSpec.unsharded(),
MemWalWriteSpec.bucket("id", 4),
MemWalWriteSpec.identity("id"),
):
with pytest.raises(Exception, match="primary key"):
table.enable_memwal(spec)
assert table.get_memwal_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.set_lsm_write_spec(LsmWriteSpec.unsharded())
table.unset_lsm_write_spec()
table.enable_memwal(MemWalWriteSpec.unsharded())
table.disable_memwal()
def test_lsm_write_spec_repr():
s = LsmWriteSpec.bucket("id", 4)
def test_memwal_spec_repr():
s = MemWalWriteSpec.bucket("id", 4)
assert s.spec_type == "bucket"
assert s.column == "id"
assert s.num_buckets == 4
@@ -95,20 +109,20 @@ def test_lsm_write_spec_repr():
assert "id" in repr(s)
assert "4" in repr(s)
u = LsmWriteSpec.unsharded()
u = MemWalWriteSpec.unsharded()
assert u.spec_type == "unsharded"
assert u.column is None
assert u.num_buckets is None
assert "unsharded" in repr(u)
def test_lsm_write_spec_with_maintained_indexes():
s = LsmWriteSpec.bucket("id", 4).with_maintained_indexes(["idx_a", "idx_b"])
def test_memwal_spec_with_maintained_indexes():
s = MemWalWriteSpec.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_unset_lsm_write_spec(tmp_path):
async def test_async_set_disable_memwal(tmp_path):
db = await lancedb.connect_async(
tmp_path, read_consistency_interval=timedelta(seconds=0)
)
@@ -118,11 +132,11 @@ async def test_async_set_unset_lsm_write_spec(tmp_path):
)
await table.set_unenforced_primary_key("id")
await table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 4))
await table.unset_lsm_write_spec()
await table.enable_memwal(MemWalWriteSpec.bucket("id", 4))
await table.disable_memwal()
# A second unset errors.
with pytest.raises(Exception, match="no LSM write spec"):
await table.unset_lsm_write_spec()
with pytest.raises(Exception, match="no MemWAL write spec"):
await table.disable_memwal()
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.set_lsm_write_spec(LsmWriteSpec.identity("v"))
table.unset_lsm_write_spec()
table.enable_memwal(MemWalWriteSpec.identity("v"))
table.disable_memwal()
def test_lsm_write_spec_identity_and_writer_config_defaults():
s = LsmWriteSpec.identity("v")
def test_memwal_spec_identity_and_writer_config_defaults():
s = MemWalWriteSpec.identity("v")
assert s.spec_type == "identity"
assert s.column == "v"
assert s.num_buckets is None
@@ -146,12 +160,12 @@ def test_lsm_write_spec_identity_and_writer_config_defaults():
assert "durable_write" in repr(s)
def test_get_lsm_write_spec(tmp_path):
def test_get_memwal_spec(tmp_path):
_db, table = _make_table(tmp_path)
table.set_unenforced_primary_key("id")
# None when nothing is installed.
assert table.get_lsm_write_spec() is None
assert table.get_memwal_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_lsm_write_spec(tmp_path):
# Bucket spec round-trips, including maintained indexes and writer config
# defaults.
table.set_lsm_write_spec(
LsmWriteSpec.bucket("id", 4)
table.enable_memwal(
MemWalWriteSpec.bucket("id", 4)
.with_maintained_indexes([idx_name])
.with_writer_config_defaults({"durable_write": "false"})
)
spec = table.get_lsm_write_spec()
spec = table.get_memwal_spec()
assert spec is not None
assert spec.spec_type == "bucket"
assert spec.column == "id"
@@ -173,30 +187,30 @@ def test_get_lsm_write_spec(tmp_path):
assert spec.writer_config_defaults == {"durable_write": "false"}
# After unset, None again.
table.unset_lsm_write_spec()
assert table.get_lsm_write_spec() is None
table.disable_memwal()
assert table.get_memwal_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.set_lsm_write_spec(LsmWriteSpec.identity("id"))
spec = table.get_lsm_write_spec()
table.enable_memwal(MemWalWriteSpec.identity("id"))
spec = table.get_memwal_spec()
assert spec.spec_type == "identity"
assert spec.column == "id"
assert spec.maintained_indexes is None
table.unset_lsm_write_spec()
table.disable_memwal()
# Unsharded round-trips (no routing column). Opting out is distinct from
# the unnamed default.
table.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
spec = table.get_lsm_write_spec()
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([]))
spec = table.get_memwal_spec()
assert spec.spec_type == "unsharded"
assert spec.column is None
assert spec.maintained_indexes == []
@pytest.mark.asyncio
async def test_async_get_lsm_write_spec(tmp_path):
async def test_async_get_memwal_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_lsm_write_spec(tmp_path):
pa.RecordBatchReader.from_batches(SCHEMA, [_batch(["seed"], [0])]),
)
assert await table.get_lsm_write_spec() is None
assert await table.get_memwal_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.set_lsm_write_spec(
LsmWriteSpec.bucket("id", 8).with_maintained_indexes([idx_name])
await table.enable_memwal(
MemWalWriteSpec.bucket("id", 8).with_maintained_indexes([idx_name])
)
spec = await table.get_lsm_write_spec()
spec = await table.get_memwal_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.unset_lsm_write_spec()
assert await table.get_lsm_write_spec() is None
await table.disable_memwal()
assert await table.get_memwal_spec() is None
python/python/tests/test_merge_insert_lsm.py
+50 -52
View File
@@ -8,7 +8,7 @@ from datetime import timedelta
import lancedb
import pyarrow as pa
import pytest
from lancedb._lancedb import LsmWriteSpec
from lancedb._lancedb import MemWalWriteSpec
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 LSM spec."""
"""A table with ``id`` as the primary key and a single-bucket MemWAL 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.set_lsm_write_spec(LsmWriteSpec.bucket("id", 1))
table.enable_memwal(MemWalWriteSpec.bucket("id", 1))
return table
@@ -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.set_lsm_write_spec(LsmWriteSpec.unsharded())
table.enable_memwal(MemWalWriteSpec.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.set_lsm_write_spec(LsmWriteSpec.identity("region"))
table.enable_memwal(MemWalWriteSpec.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_lsm_false(tmp_path):
def test_lsm_merge_insert_use_memwal_false(tmp_path):
table = _bucket_table(tmp_path) # rows id = 1, 2, 3
# use_lsm(False) opts out: the standard path runs and commits even with a spec.
# use_memwal(False) opts out: the standard path runs and commits even with a spec.
result = (
table.merge_insert("id")
.when_not_matched_insert_all()
.use_lsm(False)
.use_memwal(False)
.execute(_reader([3, 4, 5]))
)
assert result.num_inserted_rows == 2
assert table.count_rows() == 5
def test_lsm_merge_insert_use_lsm_true_without_spec_errors(tmp_path):
# A table with a primary key but no LSM write spec installed.
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.
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_lsm(True) demands MemWAL routing; without a spec it errors.
with pytest.raises(Exception, match="use_lsm"):
# use_memwal(True) demands MemWAL routing; without a spec it errors.
with pytest.raises(Exception, match="use_memwal"):
(
table.merge_insert("id")
.when_matched_update_all()
.when_not_matched_insert_all()
.use_lsm(True)
.use_memwal(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 LSM write spec installed.
# A table with a primary key but no MemWAL 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_lsm_writers()
table.close_memwal_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.set_lsm_write_spec(LsmWriteSpec.bucket("id", 1))
await table.enable_memwal(MemWalWriteSpec.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_lsm_writers()
await table.close_memwal_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.set_lsm_write_spec(LsmWriteSpec.unsharded())
table.enable_memwal(MemWalWriteSpec.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_lsm(False) bypasses the MemWAL and reads the base table only.
base_only = table.search().use_lsm(False).to_arrow()
# use_memwal(False) bypasses the MemWAL and reads the base table only.
base_only = table.search().use_memwal(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.set_lsm_write_spec(LsmWriteSpec.unsharded())
table.enable_memwal(MemWalWriteSpec.unsharded())
_lsm_upsert(table, [2, 3, 4]) # ids 2,3,4 -> values 0,1,2
@@ -258,22 +258,22 @@ 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 LSM write spec
table.set_unenforced_primary_key("id") # no MemWAL write spec
# No spec: default read and use_lsm(False) both read the base table, no error.
# No spec: default read and use_memwal(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_lsm(False).to_arrow()["id"].to_pylist()) == [
assert sorted(table.search().use_memwal(False).to_arrow()["id"].to_pylist()) == [
1,
2,
3,
]
def test_lsm_read_unsupported_shape_errors_without_use_lsm_false(tmp_path):
def test_lsm_read_unsupported_shape_errors_without_use_memwal_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.set_lsm_write_spec(LsmWriteSpec.unsharded())
table.enable_memwal(MemWalWriteSpec.unsharded())
_lsm_upsert(table, [4])
# with_row_id is unsupported by the LSM scanner; on a MemWAL table the default
@@ -281,8 +281,8 @@ def test_lsm_read_unsupported_shape_errors_without_use_lsm_false(tmp_path):
with pytest.raises(Exception):
table.search().with_row_id(True).to_arrow()
# use_lsm(False) is the escape hatch: it reads the base table only.
base = table.search().with_row_id(True).use_lsm(False).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()
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.set_lsm_write_spec(LsmWriteSpec.unsharded())
await table.enable_memwal(MemWalWriteSpec.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 LSM spec that maintains that index for the memtable.
unsharded MemWAL 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,8 +353,8 @@ 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.set_lsm_write_spec(
LsmWriteSpec.unsharded().with_maintained_indexes([index_name])
table.enable_memwal(
MemWalWriteSpec.unsharded().with_maintained_indexes([index_name])
)
return table
@@ -379,9 +379,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_lsm(False) bypasses the MemWAL, so the in-flight row is not visible.
# use_memwal(False) bypasses the MemWAL, so the in-flight row is not visible.
base_ids = set(
table.search(query).use_lsm(False).limit(100).to_arrow()["id"].to_pylist()
table.search(query).use_memwal(False).limit(100).to_arrow()["id"].to_pylist()
)
assert {1, 2, 3} <= base_ids
assert 1000 not in base_ids
@@ -453,8 +453,8 @@ 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.set_lsm_write_spec(
LsmWriteSpec.unsharded().with_maintained_indexes([index_name])
table.enable_memwal(
MemWalWriteSpec.unsharded().with_maintained_indexes([index_name])
)
# in-flight doc 4 lands in the memtable's maintained FTS index.
@@ -498,8 +498,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_lsm(False) is the escape hatch: the base-only standard path honors them.
base = table.search(query).distance_range(0.0, 100.0).use_lsm(False).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()
assert 1000 not in set(base["id"].to_pylist())
@@ -544,7 +544,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.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([]))
with pytest.raises(Exception, match="maintained"):
table.search("fox", query_type="fts", fts_columns="text").to_arrow()
@@ -553,7 +553,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.set_lsm_write_spec(LsmWriteSpec.unsharded())
table.enable_memwal(MemWalWriteSpec.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 +562,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_lsm(False) reads the base table at the pinned version.
base = table.search().use_lsm(False).to_arrow()
# use_memwal(False) reads the base table at the pinned version.
base = table.search().use_memwal(False).to_arrow()
assert sorted(base["id"].to_pylist()) == [1, 2, 3]
@@ -571,14 +571,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.set_lsm_write_spec(LsmWriteSpec.unsharded())
table.enable_memwal(MemWalWriteSpec.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_lsm(False) is the escape hatch: it reads the base table.
base = table.take_row_ids([0, 1]).use_lsm(False).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()
assert base.num_rows == 2
@@ -588,8 +588,8 @@ 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.set_lsm_write_spec(
LsmWriteSpec.unsharded().with_maintained_indexes([index_name])
table.enable_memwal(
MemWalWriteSpec.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).
@@ -608,9 +608,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.set_lsm_write_spec(
LsmWriteSpec.unsharded().with_maintained_indexes(["fts_a"])
)
table.enable_memwal(MemWalWriteSpec.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.
@@ -631,7 +629,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.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([]))
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([]))
with pytest.raises(Exception, match="maintained"):
table.search([1.0] * VECTOR_DIM).to_arrow()
@@ -644,7 +642,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.set_lsm_write_spec(LsmWriteSpec.unsharded().with_maintained_indexes([name]))
table.enable_memwal(MemWalWriteSpec.unsharded().with_maintained_indexes([name]))
# Multiple physical segments of one logical index must not be miscounted as
# multiple indexes and rejected.
ids = set(
python/python/tests/test_remote_db.py
+71 -30
View File
@@ -1623,8 +1623,6 @@ def remote_unsupported_operations():
return [
("uses_v2_manifest_paths", (), {}, "uses_v2_manifest_paths"),
("migrate_v2_manifest_paths", (), {}, "migrate_manifest_paths_v2"),
("set_unenforced_primary_key", (["id"],), {}, "set_unenforced_primary_key"),
("fetch_blob_ranges", ("image", [(1, 0, 1)]), {}, "fetch_blob_ranges"),
("optimize", (), {}, "optimize"),
("create_index", ("vector",), {"config": HnswPq()}, "IVF_HNSW_PQ"),
]
@@ -1846,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_lsm, compact_lsm, get_lsm_stats, and is
where route is one of flush_memwal, compact_memwal, get_memwal_stats, and is
responsible for writing the response.
"""
routes = ("flush_lsm", "compact_lsm", "get_lsm_stats")
routes = ("flush_memwal", "compact_memwal", "get_memwal_stats")
def handler(request):
match = re.fullmatch(r"/v1/table/test/(\w+)/", request.path)
@@ -1881,9 +1879,9 @@ def send_json(request, payload, status=200):
request.wfile.write(json.dumps(payload).encode())
def test_get_lsm_stats_sync():
def test_get_memwal_stats_sync():
"""The sync wrapper round-trips the server payload into a dict."""
bucket = {
table_shard = {
"shard_id": "b0",
"status": "Active",
"writer_epoch": 3,
@@ -1891,7 +1889,7 @@ def test_get_lsm_stats_sync():
"current_generation": 6,
"replay_after_wal_entry_position": 40,
"wal_entry_position_last_seen": 42,
"generations": [{"generation": 5, "bytes": 1024, "rows": 7}],
"sstables": [{"generation": 5, "bytes": 1024, "rows": 7}],
"compacting": False,
"memtables": [
{
@@ -1906,29 +1904,29 @@ def test_get_lsm_stats_sync():
seen_bodies = []
def lsm_handler(request, route):
assert route == "get_lsm_stats"
assert route == "get_memwal_stats"
seen_bodies.append(read_json_body(request))
send_json(request, {"lsm_stats": {"buckets": [bucket]}})
send_json(request, {"memwal_stats": {"table_shards": [table_shard]}})
with lsm_test_table(lsm_handler) as table:
assert table.get_lsm_stats() == {"buckets": [bucket]}
assert table.get_memwal_stats() == {"table_shards": [table_shard]}
# Off by default, and forwarded when asked for.
assert seen_bodies == [{"include_generation_rows": False}]
table.get_lsm_stats(include_generation_rows=True)
assert seen_bodies[-1] == {"include_generation_rows": True}
assert seen_bodies == [{"include_sstable_rows": False}]
table.get_memwal_stats(include_sstable_rows=True)
assert seen_bodies[-1] == {"include_sstable_rows": True}
def test_get_lsm_stats_sync_returns_none_when_lsm_disabled():
"""A null envelope means the LSM write path is not enabled, not an error."""
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 lsm_handler(request, route):
send_json(request, {"lsm_stats": None})
send_json(request, {"memwal_stats": None})
with lsm_test_table(lsm_handler) as table:
assert table.get_lsm_stats() is None
assert table.get_memwal_stats() is None
def test_flush_and_compact_lsm_sync():
def test_flush_and_compact_memwal_sync():
"""Both are one-shot POSTs answered 202 with no body."""
called = []
@@ -1938,13 +1936,13 @@ def test_flush_and_compact_lsm_sync():
request.end_headers()
with lsm_test_table(lsm_handler) as table:
assert table.flush_lsm() is None
assert table.compact_lsm() is None
assert called == ["flush_lsm", "compact_lsm"]
assert table.flush_memwal() is None
assert table.compact_memwal() is None
assert called == ["flush_memwal", "compact_memwal"]
def test_checkpoint_lsm_sync():
"""Seal, read the watermark, and return once L0 holds nothing.
def test_checkpoint_memwal_sync():
"""Freeze, read the watermark, and return once no SSTables remain.
The convergence loop itself is covered in Rust; this pins the sync
binding to the endpoints it drives.
@@ -1953,17 +1951,17 @@ def test_checkpoint_lsm_sync():
def lsm_handler(request, route):
called.append(route)
if route == "get_lsm_stats":
# An empty L0 yields no target watermark, so the loop is done
# after the seal without ever polling compaction.
send_json(request, {"lsm_stats": {"buckets": []}})
if route == "get_memwal_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": []}})
else:
request.send_response(202)
request.end_headers()
with lsm_test_table(lsm_handler) as table:
assert table.checkpoint_lsm() is None
assert called == ["flush_lsm", "get_lsm_stats"]
assert table.checkpoint_memwal() is None
assert called == ["flush_memwal", "get_memwal_stats"]
@contextlib.contextmanager
@@ -3363,13 +3361,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)
@@ -3401,6 +3406,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:
python/python/tests/test_wal_hybrid.py
+5 -5
View File
@@ -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_lsm_false_joins_on_row_ids():
builder = _wal_hybrid().use_lsm(False)
def test_wal_hybrid_with_use_memwal_false_joins_on_row_ids():
builder = _wal_hybrid().use_memwal(False)
builder._create_query_builders()
assert builder._pk_fusion is None
def test_wal_hybrid_refuses_an_explicit_with_row_id():
builder = _wal_hybrid().with_row_id(True)
with pytest.raises(NotImplementedError, match="use_lsm"):
with pytest.raises(NotImplementedError, match="use_memwal"):
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_lsm"):
with pytest.raises(NotImplementedError, match="use_memwal"):
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_lsm") as info:
with pytest.raises(NotImplementedError, match="use_memwal") as info:
with_pk_fallback(
run, mock.Mock(), fused=lambda: False, caller_requested_row_id=True
)
python/python/tests/utils.py
+3 -3
View File
@@ -73,10 +73,10 @@ class MockPermutationServer:
},
},
)
if path == f"/v1/table/{self.name}/get_lsm_write_spec/":
if path == f"/v1/table/{self.name}/get_memwal_spec/":
self._read_body(request)
# Null spec: this table has no LSM write path.
return self._json(request, {"lsm_write_spec": None})
# Null spec: this table has no MemWAL write path.
return self._json(request, {"memwal_spec": None})
if path == f"/v1/table/{self.name}/count_rows/":
self._read_body(request)
return self._json(request, self.num_rows)
python/src/lib.rs
+2 -2
View File
@@ -16,7 +16,7 @@ use query::{FTSQuery, HybridQuery, Query, VectorQuery};
use session::Session;
use table::{
AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken,
FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, LsmWriteSpec, MergeResult,
FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, MemWalWriteSpec, MergeResult,
PyBlobFile, 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::<LsmWriteSpec>()?;
m.add_class::<MemWalWriteSpec>()?;
m.add_class::<DeleteResult>()?;
m.add_class::<DropColumnsResult>()?;
m.add_class::<UpdateResult>()?;
python/src/query.rs
+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_lsm: Option<bool>,
pub use_memwal: 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_lsm: query_request.use_lsm,
use_memwal: query_request.use_memwal,
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_lsm: vector_query.base.use_lsm,
use_memwal: vector_query.base.use_memwal,
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_lsm(&mut self, enable: bool) {
self.inner = self.inner.clone().use_lsm(enable);
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(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_lsm(&mut self, enable: bool) {
self.inner = self.inner.clone().use_lsm(enable);
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(enable);
}
#[pyo3(signature = ())]
@@ -819,8 +819,8 @@ impl FTSQuery {
self.inner = self.inner.clone().fast_search();
}
pub fn use_lsm(&mut self, enable: bool) {
self.inner = self.inner.clone().use_lsm(enable);
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(enable);
}
pub fn with_row_id(&mut self) {
@@ -970,8 +970,8 @@ impl VectorQuery {
self.inner = self.inner.clone().fast_search();
}
pub fn use_lsm(&mut self, enable: bool) {
self.inner = self.inner.clone().use_lsm(enable);
pub fn use_memwal(&mut self, enable: bool) {
self.inner = self.inner.clone().use_memwal(enable);
}
pub fn with_row_id(&mut self) {
@@ -1178,9 +1178,9 @@ impl HybridQuery {
self.inner_fts.postfilter();
}
pub fn use_lsm(&mut self, enable: bool) {
self.inner_vec.use_lsm(enable);
self.inner_fts.use_lsm(enable);
pub fn use_memwal(&mut self, enable: bool) {
self.inner_vec.use_memwal(enable);
self.inner_fts.use_memwal(enable);
}
pub fn add_query_vector(&mut self, vector: Bound<'_, PyAny>) -> PyResult<()> {
python/src/table.rs
+70 -71
View File
@@ -39,16 +39,16 @@ use pyo3::{
mod scannable;
/// Convert `LsmStats` to a Python dict, preserving the per-bucket list.
/// Convert `MemWalStats` to a Python dict, preserving the per-table-shard list.
///
/// Deliberately not flattened to a table-level summary: a table is N
/// buckets on one node, and the per-bucket detail is the reason the
/// endpoint exists — flattening hides the single hot bucket someone opened
/// 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 lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult<Py<PyDict>> {
fn memwal_stats_to_py(py: Python<'_>, stats: &lancedb::table::MemWalStats) -> PyResult<Py<PyDict>> {
let out = PyDict::new(py);
let buckets = PyList::empty(py);
for b in &stats.buckets {
let table_shards = PyList::empty(py);
for b in &stats.table_shards {
let e = PyDict::new(py);
e.set_item("shard_id", &b.shard_id)?;
e.set_item("status", &b.status)?;
@@ -64,15 +64,15 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
b.wal_entry_position_last_seen,
)?;
let generations = PyList::empty(py);
for g in &b.generations {
let sstables = PyList::empty(py);
for g in &b.sstables {
let ge = PyDict::new(py);
ge.set_item("generation", g.generation)?;
ge.set_item("bytes", g.bytes)?;
ge.set_item("rows", g.rows)?;
generations.append(ge)?;
sstables.append(ge)?;
}
e.set_item("generations", generations)?;
e.set_item("sstables", sstables)?;
e.set_item("compacting", b.compacting)?;
e.set_item(
@@ -94,9 +94,9 @@ fn lsm_stats_to_py(py: Python<'_>, stats: &lancedb::table::LsmStats) -> PyResult
})
.transpose()?,
)?;
buckets.append(e)?;
table_shards.append(e)?;
}
out.set_item("buckets", buckets)?;
out.set_item("table_shards", table_shards)?;
Ok(out.unbind())
}
@@ -270,17 +270,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 LsmWriteSpec {
inner: lancedb::table::LsmWriteSpec,
pub struct MemWalWriteSpec {
inner: lancedb::table::MemWalWriteSpec,
}
#[pymethods]
impl LsmWriteSpec {
impl MemWalWriteSpec {
/// Hash-bucket sharding by the unenforced primary key column.
#[staticmethod]
pub fn bucket(column: String, num_buckets: u32) -> Self {
Self {
inner: lancedb::table::LsmWriteSpec::bucket(column, num_buckets),
inner: lancedb::table::MemWalWriteSpec::bucket(column, num_buckets),
}
}
@@ -294,7 +294,7 @@ impl LsmWriteSpec {
#[staticmethod]
pub fn identity(column: String) -> Self {
Self {
inner: lancedb::table::LsmWriteSpec::identity(column),
inner: lancedb::table::MemWalWriteSpec::identity(column),
}
}
@@ -303,7 +303,7 @@ impl LsmWriteSpec {
#[staticmethod]
pub fn unsharded() -> Self {
Self {
inner: lancedb::table::LsmWriteSpec::unsharded(),
inner: lancedb::table::MemWalWriteSpec::unsharded(),
}
}
@@ -327,33 +327,33 @@ impl LsmWriteSpec {
pub fn __repr__(&self) -> String {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket {
lancedb::table::MemWalWriteSpec::Bucket {
column,
num_buckets,
maintained_indexes,
writer_config_defaults,
} => format!(
"LsmWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
"MemWalWriteSpec.bucket(column={:?}, num_buckets={}, maintained_indexes={}, writer_config_defaults={:?})",
column,
num_buckets,
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
lancedb::table::LsmWriteSpec::Identity {
lancedb::table::MemWalWriteSpec::Identity {
column,
maintained_indexes,
writer_config_defaults,
} => format!(
"LsmWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
"MemWalWriteSpec.identity(column={:?}, maintained_indexes={}, writer_config_defaults={:?})",
column,
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
lancedb::table::LsmWriteSpec::Unsharded {
lancedb::table::MemWalWriteSpec::Unsharded {
maintained_indexes,
writer_config_defaults,
} => format!(
"LsmWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
"MemWalWriteSpec.unsharded(maintained_indexes={}, writer_config_defaults={:?})",
fmt_maintained(maintained_indexes),
writer_config_defaults,
),
@@ -365,9 +365,9 @@ impl LsmWriteSpec {
#[getter]
pub fn spec_type(&self) -> &'static str {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket { .. } => "bucket",
lancedb::table::LsmWriteSpec::Identity { .. } => "identity",
lancedb::table::LsmWriteSpec::Unsharded { .. } => "unsharded",
lancedb::table::MemWalWriteSpec::Bucket { .. } => "bucket",
lancedb::table::MemWalWriteSpec::Identity { .. } => "identity",
lancedb::table::MemWalWriteSpec::Unsharded { .. } => "unsharded",
}
}
@@ -375,9 +375,9 @@ impl LsmWriteSpec {
#[getter]
pub fn column(&self) -> Option<String> {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket { column, .. }
| lancedb::table::LsmWriteSpec::Identity { column, .. } => Some(column.clone()),
lancedb::table::LsmWriteSpec::Unsharded { .. } => None,
lancedb::table::MemWalWriteSpec::Bucket { column, .. }
| lancedb::table::MemWalWriteSpec::Identity { column, .. } => Some(column.clone()),
lancedb::table::MemWalWriteSpec::Unsharded { .. } => None,
}
}
@@ -385,7 +385,7 @@ impl LsmWriteSpec {
#[getter]
pub fn num_buckets(&self) -> Option<u32> {
match &self.inner {
lancedb::table::LsmWriteSpec::Bucket { num_buckets, .. } => Some(*num_buckets),
lancedb::table::MemWalWriteSpec::Bucket { num_buckets, .. } => Some(*num_buckets),
_ => None,
}
}
@@ -403,14 +403,14 @@ impl LsmWriteSpec {
}
}
impl From<LsmWriteSpec> for lancedb::table::LsmWriteSpec {
fn from(spec: LsmWriteSpec) -> Self {
impl From<MemWalWriteSpec> for lancedb::table::MemWalWriteSpec {
fn from(spec: MemWalWriteSpec) -> Self {
spec.inner
}
}
impl From<lancedb::table::LsmWriteSpec> for LsmWriteSpec {
fn from(inner: lancedb::table::LsmWriteSpec) -> Self {
impl From<lancedb::table::MemWalWriteSpec> for MemWalWriteSpec {
fn from(inner: lancedb::table::MemWalWriteSpec) -> Self {
Self { inner }
}
}
@@ -1636,8 +1636,8 @@ impl Table {
if let Some(use_index) = parameters.use_index {
builder.use_index(use_index);
}
if let Some(use_lsm) = parameters.use_lsm {
builder.use_lsm(use_lsm);
if let Some(use_memwal) = parameters.use_memwal {
builder.use_memwal(use_memwal);
}
if let Some(validate_single_shard) = parameters.validate_single_shard {
builder.validate_single_shard(validate_single_shard);
@@ -1662,29 +1662,29 @@ impl Table {
})
}
pub fn set_lsm_write_spec<'a>(
pub fn enable_memwal<'a>(
self_: PyRef<'a, Self>,
spec: LsmWriteSpec,
spec: MemWalWriteSpec,
) -> PyResult<Bound<'a, PyAny>> {
let inner = self_.inner_ref()?.clone();
let native_spec = lancedb::table::LsmWriteSpec::from(spec);
let native_spec = lancedb::table::MemWalWriteSpec::from(spec);
future_into_py(self_.py(), async move {
inner.set_lsm_write_spec(native_spec).await.infer_error()
inner.enable_memwal(native_spec).await.infer_error()
})
}
pub fn unset_lsm_write_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn disable_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.unset_lsm_write_spec().await.infer_error()
inner.disable_memwal().await.infer_error()
})
}
pub fn get_lsm_write_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn get_memwal_spec(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let spec = inner.get_lsm_write_spec().await.infer_error()?;
Ok(spec.map(LsmWriteSpec::from))
let spec = inner.get_memwal_spec().await.infer_error()?;
Ok(spec.map(MemWalWriteSpec::from))
})
}
@@ -1703,55 +1703,54 @@ impl Table {
Ok(())
}
/// Converge the table's LSM write path into its base table.
/// Converge the table's MemWAL 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_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn checkpoint_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.checkpoint_lsm().await.infer_error()
inner.checkpoint_memwal().await.infer_error()
})
}
/// Seal every bucket's active memtable into L0.
pub fn flush_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(
self_.py(),
async move { inner.flush_lsm().await.infer_error() },
)
}
/// Trigger a background L0 → base pass per bucket. Returns once the
/// passes are dispatched, not once they finish — watch `get_lsm_stats`.
pub fn compact_lsm(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
/// Freeze every table shard's active memtable into an SSTable.
pub fn flush_memwal(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.compact_lsm().await.infer_error()
inner.flush_memwal().await.infer_error()
})
}
/// Live LSM state, or `None` when the LSM write path is not enabled.
#[pyo3(signature = (include_generation_rows=false))]
pub fn get_lsm_stats(
/// 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>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.compact_memwal().await.infer_error()
})
}
/// Live MemWAL state, or `None` when the MemWAL write path is not enabled.
#[pyo3(signature = (include_sstable_rows=false))]
pub fn get_memwal_stats(
self_: PyRef<'_, Self>,
include_generation_rows: bool,
include_sstable_rows: bool,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let stats = inner
.get_lsm_stats(include_generation_rows)
.get_memwal_stats(include_sstable_rows)
.await
.infer_error()?;
Python::attach(|py| stats.map(|s| lsm_stats_to_py(py, &s)).transpose())
Python::attach(|py| stats.map(|s| memwal_stats_to_py(py, &s)).transpose())
})
}
pub fn close_lsm_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
pub fn close_memwal_writers(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
inner.close_lsm_writers().await.infer_error()
inner.close_memwal_writers().await.infer_error()
})
}
@@ -2083,7 +2082,7 @@ pub struct MergeInsertParams {
when_not_matched_by_source_condition_expr: Option<PyExpr>,
timeout: Option<std::time::Duration>,
use_index: Option<bool>,
use_lsm: Option<bool>,
use_memwal: Option<bool>,
validate_single_shard: Option<bool>,
}
python/vendor/lance-arrow/Cargo.toml
Vendored
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lance-arrow"
version = "14.0.0-beta.18"
version = "14.0.0-beta.19"
edition = "2024"
authors = ["Lance Devs <dev@lance.org>"]
license = "Apache-2.0"
python/vendor/lance-arrow/README.lancedb.md
Vendored
+2 -2
View File
@@ -1,5 +1,5 @@
This directory contains `rust/lance-arrow` from Lance tag `v14.0.0-beta.18`
(commit `f80bf1d1e59f40d249bce2539b78e41d93dee9ec`), licensed under Apache-2.0.
This directory contains `rust/lance-arrow` from Lance tag `v14.0.0-beta.19`
(commit `42e593fb4ad0e1ccc22c1fa5e3c30bfe92a5384d`), licensed under Apache-2.0.
LanceDB carries a local fix in `src/lib.rs` for sliced child validity bitmaps
until a Lance release includes the fix. The patch is selected in the root