rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-12 14:52:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge main into gatekeeper/fix-4315-1

Browse Source
This commit is contained in:
Gatefixer
2026-09-25 15:01:46 +00:00
parent 90295dd9e4 d0fb43b0b4
commit 617ee19b68
16 changed files with 1602 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+43 -43
View File
@@ -3588,8 +3588,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsst"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"rand 0.9.5",
@@ -5143,8 +5143,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
[[package]]
name = "lance"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arc-swap",
"arrow",
@@ -5216,8 +5216,8 @@ dependencies = [
[[package]]
name = "lance-arrow"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5239,7 +5239,7 @@ dependencies = [
[[package]]
name = "lance-arrow-scalar"
version = "58.0.0"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5253,7 +5253,7 @@ dependencies = [
[[package]]
name = "lance-arrow-stats"
version = "58.0.0"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5262,8 +5262,8 @@ dependencies = [
[[package]]
name = "lance-bitpacking"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrayref",
"crunchy",
@@ -5273,8 +5273,8 @@ dependencies = [
[[package]]
name = "lance-core"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5311,8 +5311,8 @@ dependencies = [
[[package]]
name = "lance-datafusion"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow",
"arrow-array",
@@ -5343,8 +5343,8 @@ dependencies = [
[[package]]
name = "lance-datagen"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow",
"arrow-array",
@@ -5361,8 +5361,8 @@ dependencies = [
[[package]]
name = "lance-derive"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"proc-macro2",
"quote",
@@ -5371,7 +5371,7 @@ dependencies = [
[[package]]
name = "lance-encoding"
version = "13.0.0-beta.12"
version = "13.0.0-beta.14"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5404,8 +5404,8 @@ dependencies = [
[[package]]
name = "lance-file"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -5437,8 +5437,8 @@ dependencies = [
[[package]]
name = "lance-geo"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"datafusion",
"geo-traits",
@@ -5452,8 +5452,8 @@ dependencies = [
[[package]]
name = "lance-index"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arc-swap",
"arrow",
@@ -5523,8 +5523,8 @@ dependencies = [
[[package]]
name = "lance-index-core"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5546,8 +5546,8 @@ dependencies = [
[[package]]
name = "lance-io"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow",
"arrow-array",
@@ -5589,8 +5589,8 @@ dependencies = [
[[package]]
name = "lance-linalg"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5604,8 +5604,8 @@ dependencies = [
[[package]]
name = "lance-namespace"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow",
"async-trait",
@@ -5619,8 +5619,8 @@ dependencies = [
[[package]]
name = "lance-namespace-impls"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow",
"arrow-ipc",
@@ -5673,8 +5673,8 @@ dependencies = [
[[package]]
name = "lance-select"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5688,8 +5688,8 @@ dependencies = [
[[package]]
name = "lance-table"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow",
"arrow-array",
@@ -5729,8 +5729,8 @@ dependencies = [
[[package]]
name = "lance-testing"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5743,8 +5743,8 @@ dependencies = [
[[package]]
name = "lance-tokenizer"
version = "13.0.0-beta.12"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
version = "13.0.0-beta.14"
source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
dependencies = [
"frostem",
"icu_segmenter",
Cargo.toml
+14 -14
View File
@@ -14,20 +14,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0"
[workspace.dependencies]
lance = { "version" = "=13.0.0-beta.12", default-features = false, "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=13.0.0-beta.12", default-features = false, "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=13.0.0-beta.12", default-features = false, "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { path = "vendor/lance-encoding", version = "=13.0.0-beta.12" }
lance-arrow = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
lance = { "version" = "=13.0.0-beta.14", default-features = false, "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=13.0.0-beta.14", default-features = false, "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=13.0.0-beta.14", default-features = false, "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { path = "vendor/lance-encoding", version = "=13.0.0-beta.14" }
lance-arrow = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
lancedb = { path = "rust/lancedb", default-features = false }
ahash = "0.8"
# Note that this one does not include pyarrow
java/pom.xml
+1 -1
View File
@@ -28,7 +28,7 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<arrow.version>15.0.0</arrow.version>
<lance-core.version>13.0.0-beta.12</lance-core.version>
<lance-core.version>13.0.0-beta.14</lance-core.version>
<spotless.skip>false</spotless.skip>
<spotless.version>2.30.0</spotless.version>
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
python/python/lancedb/_lancedb.pyi
+3
View File
@@ -514,6 +514,9 @@ class Table:
async def set_lsm_write_spec(self, spec: LsmWriteSpec) -> None: ...
async def unset_lsm_write_spec(self) -> None: ...
async def get_lsm_write_spec(self) -> Optional[LsmWriteSpec]: ...
# WAL-PK-FUSION: delete both.
def hybrid_pk_fusion_learned(self) -> bool: ...
def note_hybrid_pk_fusion(self) -> None: ...
async def checkpoint_lsm(self) -> None: ...
async def flush_lsm(self) -> None: ...
async def compact_lsm(self) -> None: ...
python/python/lancedb/_wal_hybrid.py
+282
View File
@@ -0,0 +1,282 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""WAL-PK-FUSION: hybrid search's primary-key fallback for MemWAL tables.
Hybrid search joins its vector and full-text legs on `_rowid`, which a MemWAL
table cannot supply: the fresh tier has no stable row id, so the server rejects
`with_row_id` before it plans. When that happens the legs are joined on a
surrogate `_rowid` built from the table's primary key instead.
This is a stopgap until MemWAL supports `_rowid`, and everything it needs lives
here so it can be removed in one pass:
1. Delete this file and `tests/test_wal_hybrid.py`.
2. Run `grep -rn WAL-PK-FUSION python/ rust/` and follow the note at each
marker; the Rust side has its own copy in
`rust/lancedb/src/query/wal_fusion.rs`.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Awaitable, Callable, List, Optional, Tuple, TypeVar
import pyarrow as pa
from ._blob import blob_v2_projection_sources
if TYPE_CHECKING:
from .types import QueryProjection
T = TypeVar("T")
# Lance records the unenforced primary key as field metadata, which survives the
# `describe` round trip — so a remote table can resolve the key without opening
# the dataset. Keys are bytes on a pyarrow field.
_PK_MARKER = b"lance-schema:unenforced-primary-key"
_PK_POSITION_MARKER = b"lance-schema:unenforced-primary-key:position"
_TRUTHY = {"1", "true", "on", "yes", "y"}
def _pk_columns(schema: pa.Schema) -> List[str]:
"""The unenforced primary key columns of `schema`, in key order.
Mirrors Lance's ordering: fields with an explicit 1-based position first, in
position order, then fields carrying only the legacy boolean marker, in
declaration order.
"""
keyed = []
for index, field in enumerate(schema):
metadata = field.metadata or {}
position = metadata.get(_PK_POSITION_MARKER)
if position is not None:
try:
position = int(position)
except ValueError:
position = None
if position is None:
legacy = metadata.get(_PK_MARKER)
if legacy is None or legacy.decode().lower() not in _TRUTHY:
continue
position = 0
keyed.append((position == 0, position, index, field.name))
return [name for *_, name in sorted(keyed)]
# Both layers raise this same sentence verbatim -- sophon's
# `LSM_WITH_ROW_ID_UNSUPPORTED` says so explicitly, and the OSS LSM scanner
# builds it from the same words -- so this substring survives either path and
# the wrapping each adds. It is a coupling to a message rather than a code,
# because the refusal arrives as a generic 400; if the server ever rewords it
# the fallback stops firing and hybrid on a MemWAL table errors visibly again,
# which is loud rather than silently wrong.
_WAL_ROW_ID_REFUSAL = "does not support with_row_id"
WAL_ROW_ID_UNSUPPORTED = (
"hybrid search on a MemWAL table cannot return _rowid: the fresh tier has "
"no stable row id, and the ids the fusion joins on are synthesized from "
"the primary key. Set use_lsm(False) to read the base table only (results "
"will exclude un-compacted MemWAL data)"
)
def _is_wal_row_id_refusal(error: BaseException) -> bool:
"""Whether `error` is the server declining `_rowid` because the table is
MemWAL-backed, as opposed to any other failure the query might hit."""
return _WAL_ROW_ID_REFUSAL in str(error)
def _with_pk_columns(columns, pk_columns: List[str]):
"""Add the key columns a projection is missing.
Returns the projection to use and the columns added, which the caller drops
from the output again so an injected key never reaches a caller who did not
select it.
"""
if not columns:
# Every non-system column, the key among them.
return columns, []
# `PyQueryRequest.select` reports a dynamic projection as (name, sql) pairs.
if isinstance(columns, (list, tuple)) and isinstance(columns[0], (list, tuple)):
columns = dict(columns)
missing = [c for c in pk_columns if c not in columns]
if isinstance(columns, dict):
return {**columns, **{c: c for c in missing}}, missing
return list(columns) + missing, missing
def _stamp_surrogate_row_ids(
tables: List[pa.Table], pk_columns: List[str]
) -> List[pa.Table]:
"""Give each table a `_rowid` column derived from its primary key.
The fusion only ever compares that column for equality — to deduplicate, to
group by in a reranker, and to restore the original scores — so on a MemWAL
table, where the server has no stable row id to hand out, any injective
mapping of the primary key does the same job. Ids are assigned in first-seen
order across `tables`, so passing them in fusion order (vector, then FTS)
preserves "first occurrence wins" and its tie break.
Writing it under the `_rowid` name is what lets every reranker, third-party
implementations included, run unmodified. The column never reaches the
caller.
"""
if not pk_columns:
raise ValueError(
"hybrid search on a MemWAL table needs an unenforced primary key to "
"deduplicate on, and this table declares none"
)
ids = {}
stamped = []
for table in tables:
if table.num_rows == 0:
# Both legs empty: the combined schema carries no key column, and no
# row that would need one.
row_ids = pa.array([], type=pa.uint64())
else:
missing = [c for c in pk_columns if c not in table.column_names]
if missing:
raise ValueError(
"hybrid search could not deduplicate: primary key "
f"column(s) {missing} missing from a result set with "
f"{table.num_rows} rows"
)
keys = zip(*(table.column(c).to_pylist() for c in pk_columns))
row_ids = pa.array(
[ids.setdefault(key, len(ids)) for key in keys], type=pa.uint64()
)
stamped.append(table.append_column("_rowid", row_ids))
return stamped
def pk_fusion_learned(table, use_lsm: Optional[bool]) -> bool:
"""Whether a previous hybrid query on `table` was refused `_rowid`, so this
one should go straight to the primary key.
Learned from a refusal, never probed, so asking is free.
"""
if table is None or use_lsm is False:
return False
# Only an unambiguous True forks to the key: a table type that answers the
# question loosely keeps the `_rowid` path.
return table._hybrid_pk_fusion_learned() is True
class PkFusion:
"""How the legs of a hybrid query on a MemWAL table are joined: on a
surrogate `_rowid` built from the primary key."""
def __init__(self, pk_columns: List[str]):
self.pk_columns = pk_columns
# Key columns the caller did not select, added to the projection so the
# surrogate can be built and dropped again before returning.
self.injected: List[str] = []
@classmethod
def for_query(
cls,
schema: pa.Schema,
projection: QueryProjection,
*,
caller_requested_row_id: bool,
) -> PkFusion:
"""Resolve the key for a hybrid query, refusing the shapes that need a
real `_rowid`."""
if caller_requested_row_id:
raise NotImplementedError(WAL_ROW_ID_UNSUPPORTED)
# A projected blob needs a real row id to fetch through, whatever else
# has turned `_with_row_id` on — including `return_score="all"`.
if blob_v2_projection_sources(schema, projection):
raise NotImplementedError(
"hybrid search cannot project a blob column on a MemWAL table: "
"fetching blobs needs a real _rowid, and the fresh tier has no "
"stable one. Set use_lsm(False) to read the base table only "
"(results will exclude un-compacted MemWAL data)"
)
return cls(_pk_columns(schema))
def inject(self, columns):
"""`columns` with the key columns it is missing, to project in place of
the `_rowid` a MemWAL table would refuse."""
columns, self.injected = _with_pk_columns(columns, self.pk_columns)
return columns
def stamp(
self, vector_results: pa.Table, fts_results: pa.Table
) -> Tuple[pa.Table, pa.Table]:
"""Give both legs a surrogate `_rowid` built from the key."""
# Vector first: `merge_results` concatenates in that order and keeps
# the first occurrence, so first-seen ids preserve its tie break.
vector_results, fts_results = _stamp_surrogate_row_ids(
[vector_results, fts_results], self.pk_columns
)
return vector_results, fts_results
def strip(self, results: pa.Table) -> pa.Table:
"""Drop the surrogate and any injected key columns from the reranked
results.
The surrogate is an internal join key, never an answer, and a caller
who asked for `_rowid` was refused in `for_query`.
"""
drop = [c for c in ["_rowid", *self.injected] if c in results.column_names]
return results.drop(drop) if drop else results
def _check_refusal(error: Exception, *, fused: bool, caller_requested_row_id: bool):
"""Re-raise `error` unless it is a MemWAL `_rowid` refusal worth retrying on
the primary key."""
if fused or not _is_wal_row_id_refusal(error):
raise error
# `return_score="all"` turns row ids on for the reranker, not the caller, so
# it still falls back; only a caller who asked for them is refused, and with
# the reason rather than a bare 400.
if caller_requested_row_id:
raise NotImplementedError(WAL_ROW_ID_UNSUPPORTED) from error
def with_pk_fallback(
run: Callable[[], T],
table,
*,
fused: Callable[[], bool],
caller_requested_row_id: bool,
) -> T:
"""Run a hybrid query, retrying on the primary key if the table refuses
`_rowid`.
`_rowid` is asked for first rather than paying a round trip up front to
find out: only a MemWAL table refuses, and the refusal has to be handled
anyway -- a write spec installed elsewhere can arrive between any two
queries. It costs nothing server-side: it is raised before the query is
planned. `fused` reports whether the attempt already used the key.
"""
try:
return run()
except Exception as e:
_check_refusal(
e, fused=fused(), caller_requested_row_id=caller_requested_row_id
)
# Learned, so this retry and later queries go straight to the key.
table._note_hybrid_pk_fusion()
return run()
async def with_pk_fallback_async(
run: Callable[[], Awaitable[T]],
table,
*,
fused: Callable[[], bool],
caller_requested_row_id: bool,
) -> T:
"""The async form of `with_pk_fallback`."""
try:
return await run()
except Exception as e:
_check_refusal(
e, fused=fused(), caller_requested_row_id=caller_requested_row_id
)
table._note_hybrid_pk_fusion()
return await run()
python/python/lancedb/query.py
+111 -19
View File
@@ -41,6 +41,7 @@ from .rerankers.rrf import RRFReranker
from .rerankers.util import check_reranker_result
from .schema import is_blob_like_field, schema_has_blob_field
from .util import flatten_columns
from . import _wal_hybrid # WAL-PK-FUSION: delete.
from ._blob import (
BLOB_MODE_TO_HANDLING,
FetchBlobsAsync,
@@ -2207,6 +2208,12 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
self._phrase_query = None
self._lower_bound = None
self._upper_bound = None
# WAL-PK-FUSION: delete both, and the `with_row_id` override below.
# Set by `_create_query_builders` when the legs are fused on the key.
self._pk_fusion: Optional[_wal_hybrid.PkFusion] = None
# `_with_row_id` is also turned on by `rerank(return_score="all")` for
# the reranker's own use, so it cannot answer "did the caller ask?".
self._caller_requested_row_id = False
def _validate_query(self, query, vector=None, text=None):
if query is not None and (vector is not None or text is not None):
@@ -2228,6 +2235,16 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
return vector_query, text_query
def with_row_id(self, with_row_id: bool) -> Self: # WAL-PK-FUSION: delete.
"""Set whether to return row ids.
Recorded separately from `_with_row_id`, which `rerank(return_score=
"all")` also sets for its own use — on a MemWAL table only a caller who
asked is refused, and the reranker still falls back to the primary key.
"""
self._caller_requested_row_id = with_row_id
return super().with_row_id(with_row_id)
def phrase_query(self, phrase_query: bool = True) -> LanceHybridQueryBuilder:
"""Set whether to use phrase query.
@@ -2248,16 +2265,30 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
raise NotImplementedError("to_query_object not yet supported on a hybrid query")
def to_arrow(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
# WAL-PK-FUSION: without the fallback, this body is `_run_hybrid`'s.
return _wal_hybrid.with_pk_fallback(
lambda: self._run_hybrid(timeout=timeout),
self._table,
fused=lambda: self._pk_fusion is not None,
caller_requested_row_id=self._caller_requested_row_id,
)
def _run_hybrid(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
self._create_query_builders()
fts_query, vector_query = self._fts_query, self._vector_query
# WAL-PK-FUSION: without the fallback, both legs always ask for row ids.
if self._pk_fusion is None:
fts_query = fts_query.with_row_id(True)
vector_query = vector_query.with_row_id(True)
with ThreadPoolExecutor() as executor:
fts_future = executor.submit(
self._fts_query.with_row_id(True).to_arrow, timeout=timeout
)
vector_future = executor.submit(
self._vector_query.with_row_id(True).to_arrow, timeout=timeout
)
fts_future = executor.submit(fts_query.to_arrow, timeout=timeout)
vector_future = executor.submit(vector_query.to_arrow, timeout=timeout)
fts_results = fts_future.result()
vector_results = vector_future.result()
if self._pk_fusion is not None: # WAL-PK-FUSION: delete.
vector_results, fts_results = self._pk_fusion.stamp(
vector_results, fts_results
)
results = self._combine_hybrid_results(
fts_results=fts_results,
@@ -2272,6 +2303,8 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
return self._finish_hybrid_results(results)
def _finish_hybrid_results(self, results: pa.Table) -> pa.Table:
if self._pk_fusion is not None: # WAL-PK-FUSION: delete.
return self._pk_fusion.strip(results)
if self._user_requested_row_id():
return results
if self._blob_auto_row_id_enabled():
@@ -2445,7 +2478,10 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
self._norm = normalize
self._reranker = reranker
if reranker.score == "all":
self.with_row_id(True)
# WAL-PK-FUSION: without the fallback, `self.with_row_id(True)`.
# Not through `with_row_id`, which records caller intent: the
# reranker's need for row ids must not refuse a MemWAL table.
self._with_row_id = True
return self
@@ -2719,13 +2755,24 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
sub_query_limit = self._limit + (self._offset or 0)
self._vector_query.limit(sub_query_limit)
self._fts_query.limit(sub_query_limit)
if self._columns:
self._vector_query.select(self._columns)
self._fts_query.select(self._columns)
# WAL-PK-FUSION: without the fallback, select `self._columns` as is.
self._pk_fusion = None
columns = self._columns
if _wal_hybrid.pk_fusion_learned(self._table, self._use_lsm):
self._pk_fusion = _wal_hybrid.PkFusion.for_query(
self._table.schema,
self._columns,
caller_requested_row_id=self._caller_requested_row_id,
)
columns = self._pk_fusion.inject(columns)
if columns:
self._vector_query.select(columns)
self._fts_query.select(columns)
if self._where:
self._vector_query.where(self._where, not self._postfilter)
self._fts_query.where(self._where, not self._postfilter)
if self._with_row_id:
# WAL-PK-FUSION: without the fallback, `if self._with_row_id:`.
if self._with_row_id and self._pk_fusion is None:
self._vector_query.with_row_id(True)
self._fts_query.with_row_id(True)
if self._use_lsm is not None:
@@ -3866,6 +3913,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
self._inner = inner
self._norm = "score"
self._reranker = RRFReranker()
# WAL-PK-FUSION: delete.
# Set by `_create_child_queries` when the legs are fused on the key.
self._pk_fusion: Optional[_wal_hybrid.PkFusion] = None
def rerank(
self, reranker: Reranker = RRFReranker(), normalize: str = "score"
@@ -3897,7 +3947,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
return self
def _create_child_queries(
async def _create_child_queries(
self,
) -> Tuple["AsyncFTSQuery", "AsyncVectorQuery", int, int]:
"""Build the sub-queries that make up this hybrid query.
@@ -3906,7 +3956,8 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
the plans that are reported are the plans that actually run.
Returns the two sub-queries along with the effective limit and offset of
the hybrid query itself.
the hybrid query itself. The key the legs are fused on, if any, is left
on `self._pk_fusion`.
"""
fts_query = AsyncFTSQuery(self._inner.to_fts_query(), self._table)
vec_query = AsyncVectorQuery(self._inner.to_vector_query(), self._table)
@@ -3924,8 +3975,27 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
limit = DEFAULT_HYBRID_LIMIT
offset = fts_req.offset or vec_req.offset or 0
fts_query.with_row_id()
vec_query.with_row_id()
# WAL-PK-FUSION: without the fallback, keep only the two `with_row_id`
# calls.
self._pk_fusion = None
if _wal_hybrid.pk_fusion_learned(self._table, fts_req.use_lsm):
self._pk_fusion = _wal_hybrid.PkFusion.for_query(
await self._table.schema(),
_query_request_projection(fts_req),
caller_requested_row_id=self._user_requested_row_id(),
)
if self._pk_fusion is None:
fts_query.with_row_id()
vec_query.with_row_id()
else:
# Ask for the key columns instead, and never for `_rowid`.
# `select` carries the whole projection; `select_source_columns`
# keeps only plain column references, so rebuilding from it would
# drop computed ones.
columns = self._pk_fusion.inject(fts_req.select)
if self._pk_fusion.injected:
fts_query.select(columns)
vec_query.select(columns)
# offset() pushes the offset down into both sub-queries, which would make
# each of them skip its own first `offset` rows. The window has to be
@@ -3944,7 +4014,23 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
max_batch_length: Optional[int] = None,
timeout: Optional[timedelta] = None,
) -> AsyncRecordBatchReader:
fts_query, vec_query, limit, offset = self._create_child_queries()
# WAL-PK-FUSION: without the fallback, this body is `_run_hybrid`'s.
return await _wal_hybrid.with_pk_fallback_async(
lambda: self._run_hybrid(
max_batch_length=max_batch_length, timeout=timeout
),
self._table,
fused=lambda: self._pk_fusion is not None,
caller_requested_row_id=self._user_requested_row_id(),
)
async def _run_hybrid(
self,
*,
max_batch_length: Optional[int] = None,
timeout: Optional[timedelta] = None,
) -> AsyncRecordBatchReader:
fts_query, vec_query, limit, offset = await self._create_child_queries()
req = fts_query._inner.to_query_request()
blob_auto_row_id = False
@@ -3968,6 +4054,10 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
fts_query.to_arrow(timeout=timeout),
vec_query.to_arrow(timeout=timeout),
)
if self._pk_fusion is not None: # WAL-PK-FUSION: delete.
vector_results, fts_results = self._pk_fusion.stamp(
vector_results, fts_results
)
result = LanceHybridQueryBuilder._combine_hybrid_results(
fts_results=fts_results,
@@ -3979,7 +4069,9 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
with_row_ids=True,
offset=offset,
)
if (
if self._pk_fusion is not None: # WAL-PK-FUSION: delete this branch.
result = self._pk_fusion.strip(result)
elif (
not self._user_requested_row_id()
and not blob_auto_row_id
and "_rowid" in result.column_names
@@ -4028,7 +4120,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
plan : str
""" # noqa: E501
fts_query, vec_query, _, _ = self._create_child_queries()
fts_query, vec_query, _, _ = await self._create_child_queries()
vector_plan = await vec_query.explain_plan(verbose)
fts_plan = await fts_query.explain_plan(verbose)
# Indent sub-plans under the reranker
@@ -4057,7 +4149,7 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
-------
plan : str
"""
fts_query, vec_query, _, _ = self._create_child_queries()
fts_query, vec_query, _, _ = await self._create_child_queries()
results = ["Vector Search Query:"]
results.append(await vec_query.analyze_plan(distributed_metrics))
python/python/lancedb/remote/table.py
+8
View File
@@ -1036,6 +1036,14 @@ class RemoteTable(Table):
"""Read the installed LsmWriteSpec, or ``None``."""
return LOOP.run(self._table.get_lsm_write_spec())
# WAL-PK-FUSION: delete both hooks.
def _hybrid_pk_fusion_learned(self) -> bool:
# `self._table` is the async table, which owns the Rust handle.
return self._table._hybrid_pk_fusion_learned()
def _note_hybrid_pk_fusion(self) -> None:
self._table._note_hybrid_pk_fusion()
def checkpoint_lsm(self) -> None:
"""Synchronous version of
[`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
python/python/lancedb/table.py
+21
View File
@@ -2647,6 +2647,18 @@ class Table(ABC):
to check if the table is already using the new path style.
"""
# WAL-PK-FUSION: delete both hooks, here and on AsyncTable and RemoteTable.
def _hybrid_pk_fusion_learned(self) -> bool:
"""Whether a hybrid query here has already been refused ``_rowid``.
Learned from a refusal, never probed, so asking is free. ``False`` for
table types that never refuse.
"""
return False
def _note_hybrid_pk_fusion(self) -> None:
"""Remember a ``_rowid`` refusal, so later hybrid queries skip it."""
class LanceTable(Table):
"""
@@ -5294,6 +5306,15 @@ class AsyncTable:
"""
return await self._inner.get_lsm_write_spec()
# WAL-PK-FUSION: delete both hooks.
def _hybrid_pk_fusion_learned(self) -> bool:
"""See [`Table._hybrid_pk_fusion_learned`][lancedb.table.Table]."""
return self._inner.hybrid_pk_fusion_learned()
def _note_hybrid_pk_fusion(self) -> None:
"""See [`Table._note_hybrid_pk_fusion`][lancedb.table.Table]."""
self._inner.note_hybrid_pk_fusion()
async def checkpoint_lsm(self) -> None:
"""Converge this table's LSM write path into its base table.
python/python/tests/test_wal_hybrid.py
+219
View File
@@ -0,0 +1,219 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""WAL-PK-FUSION: delete this file with `lancedb/_wal_hybrid.py`."""
from unittest import mock
import lancedb
import pyarrow as pa
import pytest
from lancedb._wal_hybrid import WAL_ROW_ID_UNSUPPORTED, with_pk_fallback
from lancedb.query import LanceHybridQueryBuilder
from lancedb.rerankers import RRFReranker
def _wal_table(schema: pa.Schema) -> mock.Mock:
"""A MemWAL-backed table a previous query has already been refused
`_rowid` on, so the fusion goes straight to the primary key."""
table = mock.Mock()
table._hybrid_pk_fusion_learned.return_value = True
table.schema = schema
return table
_PK_SCHEMA = pa.schema(
[
pa.field(
"id",
pa.string(),
metadata={b"lance-schema:unenforced-primary-key:position": b"1"},
),
pa.field("text", pa.string()),
]
)
def _wal_hybrid(**select):
builder = (
LanceHybridQueryBuilder(_wal_table(_PK_SCHEMA)).vector([0.1, 0.2]).text("puppy")
)
if "columns" in select:
builder = builder.select(select["columns"])
return builder
def test_wal_hybrid_projects_the_key_instead_of_asking_for_row_ids():
builder = _wal_hybrid(columns=["text"])
builder._create_query_builders()
assert builder._pk_fusion.pk_columns == ["id"]
assert builder._pk_fusion.injected == ["id"]
for leg in (builder._vector_query, builder._fts_query):
assert leg._columns == ["text", "id"]
# A MemWAL table rejects with_row_id before it plans.
assert leg._with_row_id is not True
def test_wal_hybrid_selecting_the_key_injects_nothing():
builder = _wal_hybrid(columns=["text", "id"])
builder._create_query_builders()
assert builder._pk_fusion.injected == []
def test_wal_hybrid_without_a_projection_leaves_it_alone():
builder = _wal_hybrid()
builder._create_query_builders()
# Select-all already returns the key.
assert builder._pk_fusion.injected == []
assert builder._vector_query._columns is None
def test_base_table_hybrid_still_joins_on_row_ids():
table = mock.Mock()
table._hybrid_pk_fusion_learned.return_value = False
builder = LanceHybridQueryBuilder(table).vector([0.1, 0.2]).text("puppy")
builder._create_query_builders()
# `to_arrow` is what turns row ids on for this path; all that matters here
# is that no key was resolved and no projection was touched.
assert builder._pk_fusion is None
assert builder._vector_query._columns is None
def test_wal_hybrid_with_use_lsm_false_joins_on_row_ids():
builder = _wal_hybrid().use_lsm(False)
builder._create_query_builders()
assert builder._pk_fusion is None
def test_wal_hybrid_refuses_an_explicit_with_row_id():
builder = _wal_hybrid().with_row_id(True)
with pytest.raises(NotImplementedError, match="use_lsm"):
builder._create_query_builders()
def test_wal_hybrid_return_score_all_is_not_refused():
"""`return_score="all"` turns row ids on for the reranker, not the caller."""
builder = _wal_hybrid().rerank(RRFReranker(return_score="all"))
builder._create_query_builders()
assert builder._pk_fusion.pk_columns == ["id"]
def test_wal_hybrid_fuses_on_the_key_and_drops_it_again():
builder = _wal_hybrid(columns=["text"])
builder._create_query_builders()
vector_results = pa.table(
{"text": ["a", "b"], "id": ["a", "b"], "_distance": [0.1, 0.2]}
)
fts_results = pa.table({"text": ["b", "c"], "id": ["b", "c"], "_score": [1.0, 2.0]})
vector_results, fts_results = builder._pk_fusion.stamp(vector_results, fts_results)
combined = LanceHybridQueryBuilder._combine_hybrid_results(
fts_results=fts_results,
vector_results=vector_results,
norm="score",
fts_query="puppy",
reranker=RRFReranker(),
limit=10,
with_row_ids=True,
)
results = builder._finish_hybrid_results(combined)
assert "_rowid" not in results.column_names, "the surrogate is internal"
assert "id" not in results.column_names, "the caller selected only `text`"
assert sorted(results.column("text").to_pylist()) == ["a", "b", "c"]
@pytest.mark.parametrize(
"row_id_first", [True, False], ids=["row_id_first", "rerank_first"]
)
def test_wal_hybrid_refuses_row_id_even_with_return_score_all(row_id_first: bool):
"""`return_score="all"` sets the same flag the caller's request does, so
asking for both must still be refused — in either call order."""
builder = _wal_hybrid()
reranker = RRFReranker(return_score="all")
if row_id_first:
builder = builder.with_row_id(True).rerank(reranker)
else:
builder = builder.rerank(reranker).with_row_id(True)
with pytest.raises(NotImplementedError, match="use_lsm"):
builder._create_query_builders()
def test_wal_hybrid_refuses_a_blob_projection_with_return_score_all():
"""A projected blob needs a real row id to fetch through, which the
reranker turning `_with_row_id` on does not provide."""
schema = pa.schema(
[
pa.field(
"id",
pa.string(),
metadata={b"lance-schema:unenforced-primary-key:position": b"1"},
),
pa.field("text", pa.string()),
lancedb.blob("blob"),
]
)
builder = (
LanceHybridQueryBuilder(_wal_table(schema))
.vector([0.1, 0.2])
.text("puppy")
.select(["blob"])
.rerank(RRFReranker(return_score="all"))
)
with pytest.raises(NotImplementedError, match="blob"):
builder._create_query_builders()
_REFUSAL = RuntimeError(
"Bad request: the MemWAL LSM scanner does not support with_row_id"
)
def test_fallback_retries_on_the_key_and_remembers_it():
table = mock.Mock()
attempts = []
def run():
attempts.append(len(attempts))
if len(attempts) == 1:
raise _REFUSAL
return "fused"
result = with_pk_fallback(
run, table, fused=lambda: False, caller_requested_row_id=False
)
assert result == "fused"
assert len(attempts) == 2
table._note_hybrid_pk_fusion.assert_called_once()
def test_fallback_explains_the_refusal_to_a_caller_who_asked_for_row_ids():
def run():
raise _REFUSAL
with pytest.raises(NotImplementedError, match="use_lsm") as info:
with_pk_fallback(
run, mock.Mock(), fused=lambda: False, caller_requested_row_id=True
)
assert str(info.value) == WAL_ROW_ID_UNSUPPORTED
@pytest.mark.parametrize(
"error, fused",
[(ValueError("unrelated"), False), (_REFUSAL, True)],
ids=["other_error", "already_fused"],
)
def test_fallback_reraises_what_it_cannot_retry(error, fused):
def run():
raise error
table = mock.Mock()
with pytest.raises(type(error)):
with_pk_fallback(run, table, fused=lambda: fused, caller_requested_row_id=False)
table._note_hybrid_pk_fusion.assert_not_called()
python/src/table.rs
+15
View File
@@ -1637,6 +1637,21 @@ impl Table {
})
}
/// Whether a hybrid query on this table has already been refused `_rowid`.
/// Learned from a previous refusal, never probed, so this is free.
///
/// WAL-PK-FUSION: delete this and `note_hybrid_pk_fusion`.
pub fn hybrid_pk_fusion_learned(self_: PyRef<'_, Self>) -> PyResult<bool> {
Ok(self_.inner_ref()?.base_table().hybrid_pk_fusion_learned())
}
/// Remember that this table refused `_rowid`, so later hybrid queries skip
/// straight to the primary-key fusion.
pub fn note_hybrid_pk_fusion(self_: PyRef<'_, Self>) -> PyResult<()> {
self_.inner_ref()?.base_table().note_hybrid_pk_fusion();
Ok(())
}
/// Converge the table's LSM write path into its base table.
///
/// Best-effort: with writes flowing, new rows may land after the last
rust/lancedb/src/query.rs
+39 -5
View File
@@ -48,6 +48,7 @@ use crate::{
};
mod hybrid;
pub(crate) mod wal_fusion; // WAL-PK-FUSION: delete with the module.
pub(crate) const DEFAULT_TOP_K: usize = 10;
@@ -1411,15 +1412,37 @@ impl VectorQuery {
pub async fn execute_hybrid(
&self,
options: QueryExecutionOptions,
) -> Result<SendableRecordBatchStream> {
// WAL-PK-FUSION: without the fallback, this body is `run_hybrid`'s,
// with its `pk_fusion` branches taken as `None`.
wal_fusion::with_pk_fallback(self, |pk_fusion| {
self.run_hybrid(pk_fusion, options.clone())
})
.await
}
async fn run_hybrid(
&self,
pk_fusion: Option<wal_fusion::PkFusion>,
options: QueryExecutionOptions,
) -> Result<SendableRecordBatchStream> {
let max_batch_length = options.max_batch_length as usize;
let internal_options = options.without_output_batch_length_limit();
// clone query and specify we want to include row IDs, which can be needed for reranking
let mut fts_query = Query::new(self.parent.clone());
fts_query.request = self.request.base.clone();
fts_query = fts_query.with_row_id();
let mut vector_query = self.clone().with_row_id();
let mut vector_query = self.clone();
// WAL-PK-FUSION: without the fallback, keep only the `None` arm.
match &pk_fusion {
None => {
fts_query = fts_query.with_row_id();
vector_query = vector_query.with_row_id();
}
Some(pk_fusion) => {
pk_fusion.prepare_leg(&mut fts_query.request.select);
pk_fusion.prepare_leg(&mut vector_query.request.base.select);
}
}
vector_query.request.base.full_text_search = None;
let (fts_results, vec_results) = try_join!(
@@ -1440,6 +1463,11 @@ impl VectorQuery {
let mut fts_results = concat_batches(&fts_schema, fts_results.iter())?;
let mut vec_results = concat_batches(&vec_schema, vec_results.iter())?;
// WAL-PK-FUSION: delete.
if let Some(pk_fusion) = &pk_fusion {
(vec_results, fts_results) = pk_fusion.stamp(vec_results, fts_results)?;
}
if matches!(self.request.base.norm, Some(NormalizeMethod::Rank)) {
vec_results = hybrid::rank(vec_results, DIST_COL, None)?;
fts_results = hybrid::rank(fts_results, SCORE_COL, None)?;
@@ -1475,8 +1503,14 @@ impl VectorQuery {
results = results.slice(0, limit);
}
if !self.request.base.with_row_id {
results = results.drop_column(ROW_ID)?;
// WAL-PK-FUSION: without the fallback, keep only the `None` arm.
match &pk_fusion {
None => {
if !self.request.base.with_row_id {
results = results.drop_column(ROW_ID)?;
}
}
Some(pk_fusion) => results = pk_fusion.strip(results)?,
}
Ok(single_batch_stream(results, max_batch_length))
rust/lancedb/src/query/wal_fusion.rs
+523
View File
@@ -0,0 +1,523 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! WAL-PK-FUSION: hybrid search's primary-key fallback for MemWAL tables.
//!
//! Hybrid search joins its vector and full-text legs on `_rowid`, which a
//! MemWAL table cannot supply: the fresh tier has no stable row id, so the
//! server rejects `with_row_id` before it plans. When that happens the legs are
//! joined on a surrogate `_rowid` built from the table's primary key instead.
//!
//! This is a stopgap until MemWAL supports `_rowid`, and everything it needs
//! lives here so it can be removed in one pass:
//!
//! 1. Delete this file and its `mod wal_fusion;` line in `query.rs`.
//! 2. Run `grep -rn WAL-PK-FUSION rust/ python/` and follow the note at each
//! marker; the Python side has its own copy in `lancedb/_wal_hybrid.py`.
use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use arrow::row::{RowConverter, SortField};
use arrow_array::{RecordBatch, UInt64Array};
use arrow_schema::{DataType, Field};
use lance::dataset::ROW_ID;
use lance_arrow::RecordBatchExt;
use lance_core::datatypes::{LANCE_UNENFORCED_PRIMARY_KEY, LANCE_UNENFORCED_PRIMARY_KEY_POSITION};
use lance_core::utils::parse::str_is_truthy;
use super::{Select, VectorQuery};
use crate::error::{Error, Result};
/// The part of the MemWAL `with_row_id` refusal that identifies it.
///
/// Both layers raise this same sentence verbatim — sophon's
/// `LSM_WITH_ROW_ID_UNSUPPORTED` says so in as many words, and the OSS LSM
/// scanner builds it from the same words — so this substring survives either
/// path and the wrapping each adds. It is a coupling to a message rather than
/// a code, because the refusal arrives as a generic 400; if the server ever
/// rewords it the fallback stops firing and hybrid on a MemWAL table goes back
/// to erroring visibly, which is loud rather than silently wrong.
const WAL_ROW_ID_REFUSAL: &str = "does not support with_row_id";
/// Whether `error` is the server declining `_rowid` because the table is
/// MemWAL-backed, as opposed to any other failure the query might hit.
fn is_wal_row_id_refusal(error: &Error) -> bool {
error.to_string().contains(WAL_ROW_ID_REFUSAL)
}
/// Why a hybrid query on a MemWAL table cannot hand back `_rowid`, raised
/// whether the refusal was anticipated or came back from the server.
fn wal_row_id_unsupported() -> Error {
Error::NotSupported {
message: "hybrid search on a MemWAL table cannot return _rowid: the fresh tier has no \
stable row id, and the ids the fusion joins on are synthesized from the \
primary key. Set use_lsm(false) to read the base table only (results will \
exclude un-compacted MemWAL data)"
.to_string(),
}
}
/// Run a hybrid query, joining its legs on the primary key instead of `_rowid`
/// when the table is MemWAL-backed.
///
/// `run(None)` must join the legs on `_rowid`; `run(Some(fusion))` must build
/// them with `fusion` instead.
///
/// `_rowid` is asked for first and the primary key used only when the server
/// refuses, rather than paying a round trip up front to find out. Only a
/// MemWAL table refuses, so probing would tax every table to learn something
/// almost none of them need — and the refusal has to be handled regardless,
/// since a write spec installed elsewhere can arrive between any two queries.
/// The refusal costs nothing server-side: it is raised before the query is
/// planned.
pub(super) async fn with_pk_fallback<T, F, Fut>(query: &VectorQuery, run: F) -> Result<T>
where
F: Fn(Option<PkFusion>) -> Fut,
Fut: Future<Output = Result<T>>,
{
// A previous query on this table was already refused, so skip straight to
// the key.
if query.request.base.use_lsm != Some(false) && query.parent.hybrid_pk_fusion_learned() {
return run(Some(PkFusion::resolve(query).await?)).await;
}
match run(None).await {
Err(e) if is_wal_row_id_refusal(&e) => {
// A caller who asked for `_rowid` gets the reason, not a bare 400
// — the same message the learned path raises up front.
if query.request.base.with_row_id {
return Err(wal_row_id_unsupported());
}
// Learned, so later queries on this table skip straight to it.
query.parent.note_hybrid_pk_fusion();
run(Some(PkFusion::resolve(query).await?)).await
}
other => other,
}
}
/// How long a table remembers that it refused `_rowid`.
///
/// Nothing fetches this — the refusal seeds it — so the bound exists only to
/// stop the memory outliving a write spec someone removed through another
/// handle. Being late to notice keeps the query on the primary key, which on a
/// base table is still a valid read; the window is what stops that lasting.
const MEMORY_TTL: Duration = Duration::from_secs(30);
/// A table's memory of having refused `_rowid`, backing
/// [`BaseTable::hybrid_pk_fusion_learned`](crate::table::BaseTable::hybrid_pk_fusion_learned)
/// for table types that keep one.
#[derive(Debug, Default)]
pub struct PkFusionMemory(Mutex<Option<Instant>>);
impl PkFusionMemory {
pub fn learned(&self) -> bool {
self.0
.lock()
.unwrap()
.is_some_and(|learned| learned.elapsed() < MEMORY_TTL)
}
pub fn note(&self) {
*self.0.lock().unwrap() = Some(Instant::now());
}
/// Forget the refusal, as when the write spec is removed through this handle.
pub fn forget(&self) {
*self.0.lock().unwrap() = None;
}
}
/// How the legs of a hybrid query on a MemWAL table are joined: on a surrogate
/// `_rowid` derived client-side from the primary key.
pub(super) struct PkFusion {
columns: Vec<String>,
/// Key columns the caller did not ask for, added to the projection so the
/// surrogate can be built and dropped again before returning.
injected: Vec<String>,
}
impl PkFusion {
/// Resolve the primary key the legs will be fused on. Needs the schema,
/// which the table has already fetched and cached for its own reasons.
async fn resolve(query: &VectorQuery) -> Result<Self> {
if query.request.base.with_row_id {
return Err(wal_row_id_unsupported());
}
let schema = query.parent.schema().await?;
let columns = pk_columns(&schema);
let injected = pk_columns_to_inject(&query.request.base.select, &columns);
Ok(Self { columns, injected })
}
/// Make a leg return the key columns, in place of the `_rowid` a MemWAL
/// table would refuse.
pub(super) fn prepare_leg(&self, select: &mut Select) {
inject_pk_columns(select, &self.columns);
}
/// Give both legs a surrogate `_rowid` built from the key.
pub(super) fn stamp(
&self,
vector: RecordBatch,
fts: RecordBatch,
) -> Result<(RecordBatch, RecordBatch)> {
// Vector first: `merge_results` concatenates in that order and keeps
// the first occurrence, so first-seen ids preserve its tie break.
let mut legs = [vector, fts];
stamp_surrogate_row_ids(&mut legs, &self.columns)?;
let [vector, fts] = legs;
Ok((vector, fts))
}
/// Drop the surrogate and any injected key columns from the reranked
/// results. The surrogate is an internal join key, never an answer: it
/// goes whether or not the caller asked for `_rowid`, and `resolve` has
/// already refused the query if they did.
pub(super) fn strip(&self, results: RecordBatch) -> Result<RecordBatch> {
let mut results = results.drop_column(ROW_ID)?;
for column in &self.injected {
// A reranker is free to reshape its output, so only drop what
// actually came back.
if results.schema().column_with_name(column).is_some() {
results = results.drop_column(column)?;
}
}
Ok(results)
}
}
/// The key columns `select` does not already produce, and so would have to be
/// added to it for the surrogate to be buildable.
fn pk_columns_to_inject(select: &Select, pk_columns: &[String]) -> Vec<String> {
let produced: Vec<&String> = match select {
// Already every non-system column, the key among them.
Select::All => return Vec::new(),
Select::Columns(columns) => columns.iter().collect(),
Select::Dynamic(pairs) => pairs.iter().map(|(name, _)| name).collect(),
Select::Expr(pairs) => pairs.iter().map(|(name, _)| name).collect(),
};
pk_columns
.iter()
.filter(|pk| !produced.contains(pk))
.cloned()
.collect()
}
/// Add the key columns `select` is missing, as identity projections where the
/// selection is expression-shaped.
fn inject_pk_columns(select: &mut Select, pk_columns: &[String]) {
let missing = pk_columns_to_inject(select, pk_columns);
if missing.is_empty() {
return;
}
match select {
Select::All => {}
Select::Columns(columns) => columns.extend(missing),
Select::Dynamic(pairs) => pairs.extend(missing.into_iter().map(|pk| (pk.clone(), pk))),
Select::Expr(pairs) => {
pairs.extend(missing.into_iter().map(|pk| {
let expr = crate::expr::col(&pk);
(pk, expr)
}));
}
}
}
/// The unenforced primary key columns declared in `schema`, in key order.
///
/// Reads the field metadata directly rather than going through
/// [`lance_core::datatypes::Schema::unenforced_primary_key`], because a remote
/// table only ever has the Arrow schema `describe` returned — there is no Lance
/// schema, and no field ids, on that side of the wire.
///
/// Ordering mirrors Lance's: fields with an explicit 1-based position first, in
/// position order, then fields carrying only the legacy boolean marker, in
/// declaration order. Only top-level fields are considered, which is every
/// field a supported key dtype can live on.
fn pk_columns(schema: &arrow_schema::Schema) -> Vec<String> {
let mut keyed: Vec<(bool, u32, usize, &str)> = schema
.fields()
.iter()
.enumerate()
.filter_map(|(idx, field)| {
let metadata = field.metadata();
let position = metadata
.get(LANCE_UNENFORCED_PRIMARY_KEY_POSITION)
.and_then(|value| value.parse::<u32>().ok())
.or_else(|| {
metadata
.get(LANCE_UNENFORCED_PRIMARY_KEY)
.filter(|value| str_is_truthy(value))
.map(|_| 0)
})?;
Some((position == 0, position, idx, field.name().as_str()))
})
.collect();
keyed.sort_unstable();
keyed
.into_iter()
.map(|(_, _, _, name)| name.to_string())
.collect()
}
/// Replace each batch's join column with a dense surrogate derived from
/// `pk_columns`, in place and in the order given.
///
/// The fusion joins its two legs on [`ROW_ID`], which a MemWAL table cannot
/// supply — the fresh tier has no stable row id. It only ever compares the
/// column for equality, though (`merge_results` dedups, the rerankers group and
/// the score restore looks up), so any injective mapping of the primary key
/// serves. Ids are assigned in first-seen order across `batches`, so passing
/// them in fusion order keeps the dedup's "first occurrence wins" and its tie
/// break byte-identical to what a real row id would have produced.
///
/// Writing it under the [`ROW_ID`] name is what lets every reranker, including
/// third-party implementations that hardcode the name, run unmodified. The
/// column never reaches the caller: the hybrid query drops it before returning.
fn stamp_surrogate_row_ids(batches: &mut [RecordBatch], pk_columns: &[String]) -> Result<()> {
if pk_columns.is_empty() {
return Err(Error::InvalidInput {
message: "hybrid search on a MemWAL table needs an unenforced primary key to \
deduplicate on, and this table declares none"
.to_string(),
});
}
let mut converter: Option<RowConverter> = None;
let mut ids: HashMap<Vec<u8>, u64> = HashMap::new();
for batch in batches.iter_mut() {
// Both legs empty: `query_schemas` synthesizes a schema carrying only
// the score and join columns, so there is no key to read — and no row
// that would need one.
let row_ids: UInt64Array = if batch.num_rows() == 0 {
UInt64Array::from(Vec::<u64>::new())
} else {
let key_columns = pk_columns
.iter()
.map(|name| {
batch
.column_by_name(name)
.cloned()
.ok_or_else(|| Error::InvalidInput {
message: format!(
"hybrid search could not deduplicate: primary key column {} \
is missing from a result set with {} rows",
name,
batch.num_rows()
),
})
})
.collect::<Result<Vec<_>>>()?;
let converter = match converter {
Some(ref converter) => converter,
None => converter.insert(RowConverter::new(
key_columns
.iter()
.map(|column| SortField::new(column.data_type().clone()))
.collect(),
)?),
};
let rows = converter.convert_columns(&key_columns)?;
let next = &mut ids;
UInt64Array::from_iter_values(rows.iter().map(|row| {
let id = next.len() as u64;
*next.entry(row.as_ref().to_vec()).or_insert(id)
}))
};
// `query_schemas` synthesizes a schema carrying `ROW_ID` when a leg came
// back empty, so the column can already be there — and it is ours to
// define either way.
let base = match batch.schema().column_with_name(ROW_ID) {
Some(_) => batch.drop_column(ROW_ID)?,
None => batch.clone(),
};
*batch = base.try_with_column(
Field::new(ROW_ID, DataType::UInt64, false),
Arc::new(row_ids),
)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use arrow::array::downcast_array;
use arrow_array::{Float32Array, StringArray};
use arrow_schema::Schema;
use lance_index::{scalar::inverted::SCORE_COL, vector::DIST_COL};
use super::super::hybrid::{empty_fts_schema, empty_vec_schema};
use super::*;
fn field(name: &str, metadata: &[(&str, &str)]) -> Field {
Field::new(name, DataType::Utf8, false).with_metadata(
metadata
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect::<HashMap<_, _>>(),
)
}
#[test]
fn pk_columns_reads_the_position_marker() {
let schema = Schema::new(vec![
field("other", &[]),
field("id", &[(LANCE_UNENFORCED_PRIMARY_KEY_POSITION, "1")]),
]);
assert_eq!(pk_columns(&schema), vec!["id".to_string()]);
}
/// Key order is the position, not the order the columns are declared in.
#[test]
fn pk_columns_orders_a_composite_key_by_position() {
let schema = Schema::new(vec![
field("id", &[(LANCE_UNENFORCED_PRIMARY_KEY_POSITION, "2")]),
field("tenant", &[(LANCE_UNENFORCED_PRIMARY_KEY_POSITION, "1")]),
]);
assert_eq!(
pk_columns(&schema),
vec!["tenant".to_string(), "id".to_string()]
);
}
/// Datasets written before the position marker carry only the boolean.
#[test]
fn pk_columns_accepts_the_legacy_boolean_marker() {
let schema = Schema::new(vec![field("id", &[(LANCE_UNENFORCED_PRIMARY_KEY, "true")])]);
assert_eq!(pk_columns(&schema), vec!["id".to_string()]);
}
/// Mirrors Lance: an explicit position sorts ahead of a legacy field.
#[test]
fn pk_columns_sorts_positioned_fields_ahead_of_legacy_ones() {
let schema = Schema::new(vec![
field("legacy", &[(LANCE_UNENFORCED_PRIMARY_KEY, "yes")]),
field(
"positioned",
&[(LANCE_UNENFORCED_PRIMARY_KEY_POSITION, "1")],
),
]);
assert_eq!(
pk_columns(&schema),
vec!["positioned".to_string(), "legacy".to_string()]
);
}
#[test]
fn pk_columns_is_empty_without_a_key() {
let schema = Schema::new(vec![
field("a", &[]),
field("b", &[(LANCE_UNENFORCED_PRIMARY_KEY, "false")]),
]);
assert!(pk_columns(&schema).is_empty());
}
fn batch(ids: Vec<&str>, score: &str) -> RecordBatch {
let schema = Arc::new(Schema::new(vec![
Arc::new(Field::new("id", DataType::Utf8, false)),
Arc::new(Field::new(score, DataType::Float32, false)),
]));
let scores = Float32Array::from(vec![0.0_f32; ids.len()]);
RecordBatch::try_new(
schema,
vec![Arc::new(StringArray::from(ids)), Arc::new(scores)],
)
.unwrap()
}
fn row_ids(batch: &RecordBatch) -> Vec<u64> {
let ids: UInt64Array = downcast_array(batch.column_by_name(ROW_ID).unwrap());
ids.values().to_vec()
}
/// The same key in both legs has to land on the same id or the fusion will
/// not dedup, and ids must ascend in fusion order so the tie break holds.
#[test]
fn test_surrogate_row_ids_are_shared_across_legs_in_first_seen_order() {
let mut batches = vec![
batch(vec!["a", "b", "c"], DIST_COL),
batch(vec!["b", "d"], SCORE_COL),
];
stamp_surrogate_row_ids(&mut batches, &["id".to_string()]).unwrap();
assert_eq!(row_ids(&batches[0]), vec![0, 1, 2]);
assert_eq!(
row_ids(&batches[1]),
vec![1, 3],
"b keeps the id the vector leg gave it"
);
}
/// A string key is the common case and cannot be cast to the u64 the
/// rerankers read, which is the whole reason for the surrogate.
#[test]
fn test_surrogate_row_ids_repeat_within_a_leg() {
let mut batches = vec![batch(vec!["a", "a", "b"], DIST_COL)];
stamp_surrogate_row_ids(&mut batches, &["id".to_string()]).unwrap();
assert_eq!(row_ids(&batches[0]), vec![0, 0, 1]);
}
#[test]
fn test_surrogate_row_ids_support_a_composite_key() {
let schema = Arc::new(Schema::new(vec![
Arc::new(Field::new("tenant", DataType::Utf8, false)),
Arc::new(Field::new("id", DataType::Int32, false)),
]));
let make = |tenants: Vec<&str>, ids: Vec<i32>| {
RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(StringArray::from(tenants)),
Arc::new(arrow_array::Int32Array::from(ids)),
],
)
.unwrap()
};
let mut batches = vec![make(vec!["x", "y"], vec![1, 1]), make(vec!["y"], vec![1])];
stamp_surrogate_row_ids(&mut batches, &["tenant".to_string(), "id".to_string()]).unwrap();
assert_eq!(
row_ids(&batches[0]),
vec![0, 1],
"same id, different tenant"
);
assert_eq!(row_ids(&batches[1]), vec![1]);
}
/// Both legs empty: `query_schemas` synthesizes a schema with no key column,
/// and there is no row that would need one.
#[test]
fn test_surrogate_row_ids_tolerate_the_both_empty_schema() {
let mut batches = vec![
RecordBatch::new_empty(Arc::new(empty_vec_schema())),
RecordBatch::new_empty(Arc::new(empty_fts_schema())),
];
stamp_surrogate_row_ids(&mut batches, &["id".to_string()]).unwrap();
assert_eq!(row_ids(&batches[0]), Vec::<u64>::new());
}
/// Rows with no key would silently fuse into one bucket, so refuse.
#[test]
fn test_surrogate_row_ids_reject_a_missing_key_column() {
let mut batches = vec![batch(vec!["a"], DIST_COL)];
let err = stamp_surrogate_row_ids(&mut batches, &["nope".to_string()]).unwrap_err();
assert!(err.to_string().contains("nope"), "{err}");
}
#[test]
fn test_surrogate_row_ids_reject_a_table_with_no_primary_key() {
let mut batches = vec![batch(vec!["a"], DIST_COL)];
let err = stamp_surrogate_row_ids(&mut batches, &[]).unwrap_err();
assert!(err.to_string().contains("unenforced primary key"), "{err}");
}
}
rust/lancedb/src/remote/table.rs
+292
View File
@@ -20,6 +20,7 @@ use crate::job::Job;
use crate::materialized_view::{
MaterializedViewDefinition, MaterializedViewInfo, RefreshMaterializedViewResult, ViewProjection,
};
use crate::query::wal_fusion::PkFusionMemory; // WAL-PK-FUSION: delete.
use crate::query::{QueryFilter, QueryRequest, Select, VectorQueryRequest};
use crate::remote::job::RemoteJob;
use crate::table::AddColumnsResult;
@@ -497,6 +498,8 @@ pub struct RemoteTable<S: HttpSend = Sender> {
version: Arc<RwLock<Option<u64>>>,
location: RwLock<Option<String>>,
schema_cache: BackgroundCache<SchemaRef, Error>,
// WAL-PK-FUSION: delete, with its initializers below.
wal_pk_fusion: PkFusionMemory,
freshness: Arc<Mutex<FreshnessState>>,
/// The branch this handle is scoped to, or `None` for the main branch.
/// Stamped onto every branch-accepting request so reads and writes resolve
@@ -665,6 +668,7 @@ impl<S: HttpSend> RemoteTable<S> {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: PkFusionMemory::default(),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -698,6 +702,7 @@ impl<S: HttpSend> RemoteTable<S> {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: PkFusionMemory::default(),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch,
}
@@ -1615,6 +1620,7 @@ mod test_utils {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: PkFusionMemory::default(),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -1639,6 +1645,7 @@ mod test_utils {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: PkFusionMemory::default(),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -1672,6 +1679,7 @@ mod test_utils {
version: Arc::new(RwLock::new(None)),
location: RwLock::new(None),
schema_cache: BackgroundCache::new(SCHEMA_CACHE_TTL, SCHEMA_CACHE_REFRESH_WINDOW),
wal_pk_fusion: PkFusionMemory::default(),
freshness: Arc::new(Mutex::new(FreshnessState::default())),
branch: None,
}
@@ -3233,6 +3241,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
));
let (request_id, response) = self.send(request, true).await?;
self.check_table_response(&request_id, response).await?;
self.wal_pk_fusion.forget(); // WAL-PK-FUSION: delete.
Ok(())
}
@@ -3299,6 +3308,15 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
Ok(Some(spec))
}
// WAL-PK-FUSION: delete both.
fn hybrid_pk_fusion_learned(&self) -> bool {
self.wal_pk_fusion.learned()
}
fn note_hybrid_pk_fusion(&self) {
self.wal_pk_fusion.note();
}
async fn tags(&self) -> Result<Box<dyn Tags + '_>> {
Ok(Box::new(RemoteTags { inner: self }))
}
@@ -13585,4 +13603,278 @@ mod tests {
.unwrap();
branch.stats().await.unwrap();
}
// WAL-PK-FUSION: delete everything from here to the end of `mod tests`.
/// Verbatim from sophon's `LSM_WITH_ROW_ID_UNSUPPORTED`; the fallback keys
/// off this sentence, so a test that paraphrased it would prove nothing.
const WAL_ROW_ID_REFUSAL_BODY: &str = r#"{"code":13,"error":"Bad request: the MemWAL LSM scanner does not support with_row_id (the LSM scanner exposes _rowaddr, not a stable _rowid); set use_lsm(false) to read the base table only (results will exclude un-compacted MemWAL data)"}"#;
use lance::dataset::ROW_ID;
fn pk_schema() -> Schema {
use lance_core::datatypes::LANCE_UNENFORCED_PRIMARY_KEY_POSITION;
Schema::new(vec![
Field::new("id", DataType::Utf8, false).with_metadata(
[(
LANCE_UNENFORCED_PRIMARY_KEY_POSITION.to_string(),
"1".to_string(),
)]
.into_iter()
.collect(),
),
Field::new("text", DataType::Utf8, true),
])
}
/// One leg's worth of results, keyed so the two legs share a row.
/// `row_ids` mirrors a server answering a query that asked for `_rowid`.
fn leg(score_column: &str, ids: Vec<&str>, row_ids: bool) -> RecordBatch {
use arrow_array::{Float32Array, UInt64Array};
let mut fields = vec![
Field::new("id", DataType::Utf8, false),
Field::new("text", DataType::Utf8, true),
Field::new(score_column, DataType::Float32, false),
];
let scores = Float32Array::from(vec![1.0_f32; ids.len()]);
let texts = StringArray::from(ids.clone());
let mut columns: Vec<arrow_array::ArrayRef> = vec![
Arc::new(StringArray::from(ids.clone())),
Arc::new(texts),
Arc::new(scores),
];
if row_ids {
fields.push(Field::new(ROW_ID, DataType::UInt64, false));
columns.push(Arc::new(UInt64Array::from_iter_values(
ids.iter().map(|id| id.as_bytes()[0] as u64),
)));
}
RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap()
}
/// A server that refuses `_rowid` the way a MemWAL table does, recording
/// every query body so a test can see which key each attempt asked for.
fn wal_refusing_table(bodies: Arc<std::sync::Mutex<Vec<serde_json::Value>>>) -> Table {
Table::new_with_handler("my_table", move |request| match request.url().path() {
"/v1/table/my_table/describe/" => http::Response::builder()
.status(200)
.body(describe_response(&pk_schema()).into_bytes())
.unwrap(),
"/v1/table/my_table/query/" => {
let body = request_body_json(&request);
let asked_for_row_id = body["with_row_id"] == serde_json::Value::Bool(true);
let is_fts = body.get("full_text_query").is_some_and(|v| !v.is_null());
bodies.lock().unwrap().push(body);
if asked_for_row_id {
return http::Response::builder()
.status(400)
.body(WAL_ROW_ID_REFUSAL_BODY.as_bytes().to_vec())
.unwrap();
}
let batch = if is_fts {
leg("_score", vec!["a", "b"], false)
} else {
leg("_distance", vec!["a"], false)
};
http::Response::builder()
.status(200)
.body(write_ipc_file(&batch))
.unwrap()
}
path => panic!("unexpected request path: {path}"),
})
}
fn hybrid(table: &Table) -> crate::query::VectorQuery {
table
.query()
.full_text_search(FullTextSearchQuery::new("a".to_string()))
.nearest_to(&[0.0, 0.0])
.unwrap()
}
/// The optimistic path: ask for `_rowid`, and when a MemWAL table refuses,
/// fall back to the primary key rather than failing.
#[tokio::test]
async fn test_hybrid_falls_back_to_the_primary_key_when_row_id_is_refused() {
let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
let table = wal_refusing_table(bodies.clone());
let results = hybrid(&table)
.select(Select::columns(&["text"]))
.limit(10)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
let bodies = bodies.lock().unwrap();
let (asked_row_id, asked_pk): (Vec<_>, Vec<_>) = bodies
.iter()
.partition(|b| b["with_row_id"] == serde_json::Value::Bool(true));
// The legs run concurrently under `try_join!`, which abandons the
// sibling as soon as one errors — so the refused attempt is one or two
// requests, never zero, and the retry always sends both.
assert!(
!asked_row_id.is_empty(),
"`_rowid` has to be tried before the fallback means anything"
);
assert_eq!(asked_pk.len(), 2, "both legs are retried on the key");
for body in asked_pk {
let columns = body["columns"].as_array().expect("a column projection");
assert!(
columns.iter().any(|c| c == "id"),
"the retry has to project the key: {body}"
);
}
let batch = &results[0];
assert!(batch.column_by_name(ROW_ID).is_none());
assert!(batch.column_by_name("id").is_none(), "injected key leaked");
assert_eq!(batch.num_rows(), 2, "`a` is in both legs and fuses");
}
/// A hybrid query that matches nothing must return nothing, not fail.
/// When both legs come back empty, `query_schemas` synthesizes a schema
/// that already carries `_rowid`, which the stamping has to tolerate.
#[tokio::test]
async fn test_hybrid_with_no_matches_returns_an_empty_result() {
let table = Table::new_with_handler("my_table", move |request| {
match request.url().path() {
"/v1/table/my_table/describe/" => http::Response::builder()
.status(200)
.body(describe_response(&pk_schema()).into_bytes())
.unwrap(),
"/v1/table/my_table/query/" => {
let body = request_body_json(&request);
if body["with_row_id"] == serde_json::Value::Bool(true) {
return http::Response::builder()
.status(400)
.body(WAL_ROW_ID_REFUSAL_BODY.as_bytes().to_vec())
.unwrap();
}
// A leg that matched nothing: schema, no batches.
let schema =
Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8, false)]));
let mut body = Vec::new();
{
let mut writer =
arrow_ipc::writer::FileWriter::try_new(&mut body, &schema).unwrap();
writer.finish().unwrap();
}
http::Response::builder().status(200).body(body).unwrap()
}
path => panic!("unexpected request path: {path}"),
}
});
let results = hybrid(&table)
.limit(10)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
assert_eq!(results.iter().map(|b| b.num_rows()).sum::<usize>(), 0);
}
/// The refusal is paid once: the second query goes straight to the key.
#[tokio::test]
async fn test_hybrid_remembers_the_refusal() {
let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
let table = wal_refusing_table(bodies.clone());
for _ in 0..2 {
hybrid(&table)
.limit(10)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
}
let refused = bodies
.lock()
.unwrap()
.iter()
.filter(|b| b["with_row_id"] == serde_json::Value::Bool(true))
.count();
assert!(
(1..=2).contains(&refused),
"only the first query should pay the refusal, saw {refused} attempts"
);
}
/// A table that never refuses must never pay for the fallback existing.
#[tokio::test]
async fn test_hybrid_on_a_base_table_sends_no_extra_requests() {
let paths = Arc::new(std::sync::Mutex::new(Vec::new()));
let seen = paths.clone();
let table = Table::new_with_handler("my_table", move |request| {
seen.lock().unwrap().push(request.url().path().to_string());
let batch = if request_body_json(&request)
.get("full_text_query")
.is_some_and(|v| !v.is_null())
{
leg("_score", vec!["a", "b"], true)
} else {
leg("_distance", vec!["a"], true)
};
http::Response::builder()
.status(200)
.body(write_ipc_file(&batch))
.unwrap()
});
hybrid(&table)
.limit(10)
.execute()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.unwrap();
let paths = paths.lock().unwrap();
assert!(
paths.iter().all(|p| p.ends_with("/query/")),
"a table that never refuses must not be probed, got {paths:?}"
);
assert_eq!(paths.len(), 2, "one request per leg, got {paths:?}");
}
/// A caller who asked for `_rowid` gets the server's refusal, not a
/// surrogate dressed up as one.
#[tokio::test]
async fn test_hybrid_does_not_fall_back_when_the_caller_asked_for_row_id() {
let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
let table = wal_refusing_table(bodies.clone());
let err = hybrid(&table)
.with_row_id()
.limit(10)
.execute()
.await
.err()
.expect("the refusal must reach the caller");
// Translated, not passed through: the caller gets the reason and the
// way out, which a bare 400 does not carry.
assert!(matches!(err, Error::NotSupported { .. }), "{err:?}");
assert!(err.to_string().contains("use_lsm(false)"), "{err}");
assert!(
bodies
.lock()
.unwrap()
.iter()
.all(|b| b["with_row_id"] == serde_json::Value::Bool(true)),
"no retry should have been attempted"
);
}
}
rust/lancedb/src/table.rs
+25
View File
@@ -706,6 +706,31 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
message: "get_lsm_write_spec is not supported on this table type".into(),
})
}
/// Whether a hybrid query on this table has already been told it cannot
/// join its legs on `_rowid`.
///
/// WAL-PK-FUSION: delete this and `note_hybrid_pk_fusion`.
///
/// Learned, never probed: hybrid optimistically asks for `_rowid` and only
/// a MemWAL table refuses, so paying a round trip up front would tax every
/// table to discover something almost none of them need. Synchronous and
/// free by construction — an implementation may only answer from what a
/// previous query already learned.
///
/// The default is `false`, which keeps a table type that never refuses on
/// the `_rowid` path forever.
fn hybrid_pk_fusion_learned(&self) -> bool {
false
}
/// Record that this table refused `_rowid`, so later hybrid queries skip
/// straight to the primary-key fusion instead of paying the refusal again.
///
/// Implementations should expire this the way they expire other table
/// metadata: a spec can be removed, after which `_rowid` works again and
/// the only cost of being late to notice is a base-only read that is still
/// correct.
fn note_hybrid_pk_fusion(&self) {}
/// Seal every bucket's active memtable into L0.
///
/// The default implementation returns `NotSupported`.
vendor/lance-encoding/Cargo.toml
Vendored
+5 -5
View File
@@ -1,6 +1,6 @@
[package]
name = "lance-encoding"
version = "13.0.0-beta.12"
version = "13.0.0-beta.14"
edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/lance-format/lance"
@@ -13,10 +13,10 @@ rust-version = "1.91.0"
[workspace]
[dependencies]
lance-arrow = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12" }
lance-core = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12" }
lance-bitpacking = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12", optional = true }
fsst = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12" }
lance-arrow = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14" }
lance-core = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14" }
lance-bitpacking = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14", optional = true }
fsst = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14" }
arrow-arith = "58.0.0"
arrow-array = "58.0.0"
arrow-buffer = "58.0.0"
vendor/lance-encoding/PATCH.md
Vendored
+1 -1
View File
@@ -1,5 +1,5 @@
This is `rust/lance-encoding` from
[`lance-format/lance` v13.0.0-beta.12](https://github.com/lance-format/lance/tree/5d2946fa452c61cb018144f3200892b6dd4a8b4f/rust/lance-encoding),
[`lance-format/lance` v13.0.0-beta.14](https://github.com/lance-format/lance/tree/b80ac1d9f3c24b47bd8e297d03053ea35d216a9f/rust/lance-encoding),
with the crate's build protos copied into `protos/`. Its manifest uses explicit
dependencies so it can be built outside the Lance workspace.