From 22165860231231d738c246ccd4d0ef46f5baedd8 Mon Sep 17 00:00:00 2001
From: Dan Rammer <dan@rammer.me>
Date: Thu, 24 Sep 2026 22:02:04 -0500
Subject: [PATCH 1/2] feat(query): dedup hybrid search by primary key on MemWAL
 tables (#4268)

Implementing a PK-based dedup to support WAL in hybrid search.
Basically, the existing code just calls a vector query + FTS query and
dedups locally using the row ID. Since row ID is unsupported in WAL
(currently), we update the algorithm to use the tables PK (which is
required in WAL, and uniquely linked to a PK on updates). To enable
PK-dedup branch, we've chosen to fallback on error rather than fire an
exploratory query. Specifically, we issue the regular hybrid query,
using row IDs, and when that fails with "WAL does not support row IDs"
we fallback to the PK-based variant, and then cache the fallback for
future calls from the same client. The alternative is that we chec if
WAL is enabled before issuing the query. This would require a
`get_lsm_write_spec` call, slowing down all the existing non-WAL paths
unnecessarily.

The ultimate goal is to get row ID supported in WAL, but that is pending
stable row ID GA and subsequent integration into the enterprise stack.
Once that lands, we will rip out the PK-based dedup that is only for
WAL-enabled tables and use the existing row ID code. This is the main
reason for building this adaptively. I had spec-ed out a refactor so
that rather than duplicating code between python and rust we build this
in rust and have python call that through PyO3. However, that made the
reverse path (ie. ) more difficult. So I think the tradeoff of
maintaining duplicate code in the short term for the ease of reverting
this in the medium term (hopefully 2-3 months).

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
---
 python/python/lancedb/_lancedb.pyi     |   3 +
 python/python/lancedb/_wal_hybrid.py   | 282 +++++++++++++
 python/python/lancedb/query.py         | 130 +++++-
 python/python/lancedb/remote/table.py  |   8 +
 python/python/lancedb/table.py         |  21 +
 python/python/tests/test_wal_hybrid.py | 219 +++++++++++
 python/src/table.rs                    |  15 +
 rust/lancedb/src/query.rs              |  44 ++-
 rust/lancedb/src/query/wal_fusion.rs   | 523 +++++++++++++++++++++++++
 rust/lancedb/src/remote/table.rs       | 292 ++++++++++++++
 rust/lancedb/src/table.rs              |  25 ++
 11 files changed, 1538 insertions(+), 24 deletions(-)
 create mode 100644 python/python/lancedb/_wal_hybrid.py
 create mode 100644 python/python/tests/test_wal_hybrid.py
 create mode 100644 rust/lancedb/src/query/wal_fusion.rs

diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi
index 337b736ce..1f26ad6f4 100644
--- a/python/python/lancedb/_lancedb.pyi
+++ b/python/python/lancedb/_lancedb.pyi
@@ -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: ...
diff --git a/python/python/lancedb/_wal_hybrid.py b/python/python/lancedb/_wal_hybrid.py
new file mode 100644
index 000000000..67597950d
--- /dev/null
+++ b/python/python/lancedb/_wal_hybrid.py
@@ -0,0 +1,282 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""WAL-PK-FUSION: hybrid search's primary-key fallback for MemWAL tables.
+
+Hybrid search joins its vector and full-text legs on `_rowid`, which a MemWAL
+table cannot supply: the fresh tier has no stable row id, so the server rejects
+`with_row_id` before it plans. When that happens the legs are joined on a
+surrogate `_rowid` built from the table's primary key instead.
+
+This is a stopgap until MemWAL supports `_rowid`, and everything it needs lives
+here so it can be removed in one pass:
+
+1. Delete this file and `tests/test_wal_hybrid.py`.
+2. Run `grep -rn WAL-PK-FUSION python/ rust/` and follow the note at each
+   marker; the Rust side has its own copy in
+   `rust/lancedb/src/query/wal_fusion.rs`.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Awaitable, Callable, List, Optional, Tuple, TypeVar
+
+import pyarrow as pa
+
+from ._blob import blob_v2_projection_sources
+
+if TYPE_CHECKING:
+    from .types import QueryProjection
+
+T = TypeVar("T")
+
+
+# Lance records the unenforced primary key as field metadata, which survives the
+# `describe` round trip — so a remote table can resolve the key without opening
+# the dataset. Keys are bytes on a pyarrow field.
+_PK_MARKER = b"lance-schema:unenforced-primary-key"
+_PK_POSITION_MARKER = b"lance-schema:unenforced-primary-key:position"
+_TRUTHY = {"1", "true", "on", "yes", "y"}
+
+
+def _pk_columns(schema: pa.Schema) -> List[str]:
+    """The unenforced primary key columns of `schema`, in key order.
+
+    Mirrors Lance's ordering: fields with an explicit 1-based position first, in
+    position order, then fields carrying only the legacy boolean marker, in
+    declaration order.
+    """
+    keyed = []
+    for index, field in enumerate(schema):
+        metadata = field.metadata or {}
+        position = metadata.get(_PK_POSITION_MARKER)
+        if position is not None:
+            try:
+                position = int(position)
+            except ValueError:
+                position = None
+        if position is None:
+            legacy = metadata.get(_PK_MARKER)
+            if legacy is None or legacy.decode().lower() not in _TRUTHY:
+                continue
+            position = 0
+        keyed.append((position == 0, position, index, field.name))
+    return [name for *_, name in sorted(keyed)]
+
+
+# Both layers raise this same sentence verbatim -- sophon's
+# `LSM_WITH_ROW_ID_UNSUPPORTED` says so explicitly, and the OSS LSM scanner
+# builds it from the same words -- so this substring survives either path and
+# the wrapping each adds. It is a coupling to a message rather than a code,
+# because the refusal arrives as a generic 400; if the server ever rewords it
+# the fallback stops firing and hybrid on a MemWAL table errors visibly again,
+# which is loud rather than silently wrong.
+_WAL_ROW_ID_REFUSAL = "does not support with_row_id"
+
+WAL_ROW_ID_UNSUPPORTED = (
+    "hybrid search on a MemWAL table cannot return _rowid: the fresh tier has "
+    "no stable row id, and the ids the fusion joins on are synthesized from "
+    "the primary key. Set use_lsm(False) to read the base table only (results "
+    "will exclude un-compacted MemWAL data)"
+)
+
+
+def _is_wal_row_id_refusal(error: BaseException) -> bool:
+    """Whether `error` is the server declining `_rowid` because the table is
+    MemWAL-backed, as opposed to any other failure the query might hit."""
+    return _WAL_ROW_ID_REFUSAL in str(error)
+
+
+def _with_pk_columns(columns, pk_columns: List[str]):
+    """Add the key columns a projection is missing.
+
+    Returns the projection to use and the columns added, which the caller drops
+    from the output again so an injected key never reaches a caller who did not
+    select it.
+    """
+    if not columns:
+        # Every non-system column, the key among them.
+        return columns, []
+    # `PyQueryRequest.select` reports a dynamic projection as (name, sql) pairs.
+    if isinstance(columns, (list, tuple)) and isinstance(columns[0], (list, tuple)):
+        columns = dict(columns)
+    missing = [c for c in pk_columns if c not in columns]
+    if isinstance(columns, dict):
+        return {**columns, **{c: c for c in missing}}, missing
+    return list(columns) + missing, missing
+
+
+def _stamp_surrogate_row_ids(
+    tables: List[pa.Table], pk_columns: List[str]
+) -> List[pa.Table]:
+    """Give each table a `_rowid` column derived from its primary key.
+
+    The fusion only ever compares that column for equality — to deduplicate, to
+    group by in a reranker, and to restore the original scores — so on a MemWAL
+    table, where the server has no stable row id to hand out, any injective
+    mapping of the primary key does the same job. Ids are assigned in first-seen
+    order across `tables`, so passing them in fusion order (vector, then FTS)
+    preserves "first occurrence wins" and its tie break.
+
+    Writing it under the `_rowid` name is what lets every reranker, third-party
+    implementations included, run unmodified. The column never reaches the
+    caller.
+    """
+    if not pk_columns:
+        raise ValueError(
+            "hybrid search on a MemWAL table needs an unenforced primary key to "
+            "deduplicate on, and this table declares none"
+        )
+
+    ids = {}
+    stamped = []
+    for table in tables:
+        if table.num_rows == 0:
+            # Both legs empty: the combined schema carries no key column, and no
+            # row that would need one.
+            row_ids = pa.array([], type=pa.uint64())
+        else:
+            missing = [c for c in pk_columns if c not in table.column_names]
+            if missing:
+                raise ValueError(
+                    "hybrid search could not deduplicate: primary key "
+                    f"column(s) {missing} missing from a result set with "
+                    f"{table.num_rows} rows"
+                )
+            keys = zip(*(table.column(c).to_pylist() for c in pk_columns))
+            row_ids = pa.array(
+                [ids.setdefault(key, len(ids)) for key in keys], type=pa.uint64()
+            )
+        stamped.append(table.append_column("_rowid", row_ids))
+    return stamped
+
+
+def pk_fusion_learned(table, use_lsm: Optional[bool]) -> bool:
+    """Whether a previous hybrid query on `table` was refused `_rowid`, so this
+    one should go straight to the primary key.
+
+    Learned from a refusal, never probed, so asking is free.
+    """
+    if table is None or use_lsm is False:
+        return False
+    # Only an unambiguous True forks to the key: a table type that answers the
+    # question loosely keeps the `_rowid` path.
+    return table._hybrid_pk_fusion_learned() is True
+
+
+class PkFusion:
+    """How the legs of a hybrid query on a MemWAL table are joined: on a
+    surrogate `_rowid` built from the primary key."""
+
+    def __init__(self, pk_columns: List[str]):
+        self.pk_columns = pk_columns
+        # Key columns the caller did not select, added to the projection so the
+        # surrogate can be built and dropped again before returning.
+        self.injected: List[str] = []
+
+    @classmethod
+    def for_query(
+        cls,
+        schema: pa.Schema,
+        projection: QueryProjection,
+        *,
+        caller_requested_row_id: bool,
+    ) -> PkFusion:
+        """Resolve the key for a hybrid query, refusing the shapes that need a
+        real `_rowid`."""
+        if caller_requested_row_id:
+            raise NotImplementedError(WAL_ROW_ID_UNSUPPORTED)
+        # A projected blob needs a real row id to fetch through, whatever else
+        # has turned `_with_row_id` on — including `return_score="all"`.
+        if blob_v2_projection_sources(schema, projection):
+            raise NotImplementedError(
+                "hybrid search cannot project a blob column on a MemWAL table: "
+                "fetching blobs needs a real _rowid, and the fresh tier has no "
+                "stable one. Set use_lsm(False) to read the base table only "
+                "(results will exclude un-compacted MemWAL data)"
+            )
+        return cls(_pk_columns(schema))
+
+    def inject(self, columns):
+        """`columns` with the key columns it is missing, to project in place of
+        the `_rowid` a MemWAL table would refuse."""
+        columns, self.injected = _with_pk_columns(columns, self.pk_columns)
+        return columns
+
+    def stamp(
+        self, vector_results: pa.Table, fts_results: pa.Table
+    ) -> Tuple[pa.Table, pa.Table]:
+        """Give both legs a surrogate `_rowid` built from the key."""
+        # Vector first: `merge_results` concatenates in that order and keeps
+        # the first occurrence, so first-seen ids preserve its tie break.
+        vector_results, fts_results = _stamp_surrogate_row_ids(
+            [vector_results, fts_results], self.pk_columns
+        )
+        return vector_results, fts_results
+
+    def strip(self, results: pa.Table) -> pa.Table:
+        """Drop the surrogate and any injected key columns from the reranked
+        results.
+
+        The surrogate is an internal join key, never an answer, and a caller
+        who asked for `_rowid` was refused in `for_query`.
+        """
+        drop = [c for c in ["_rowid", *self.injected] if c in results.column_names]
+        return results.drop(drop) if drop else results
+
+
+def _check_refusal(error: Exception, *, fused: bool, caller_requested_row_id: bool):
+    """Re-raise `error` unless it is a MemWAL `_rowid` refusal worth retrying on
+    the primary key."""
+    if fused or not _is_wal_row_id_refusal(error):
+        raise error
+    # `return_score="all"` turns row ids on for the reranker, not the caller, so
+    # it still falls back; only a caller who asked for them is refused, and with
+    # the reason rather than a bare 400.
+    if caller_requested_row_id:
+        raise NotImplementedError(WAL_ROW_ID_UNSUPPORTED) from error
+
+
+def with_pk_fallback(
+    run: Callable[[], T],
+    table,
+    *,
+    fused: Callable[[], bool],
+    caller_requested_row_id: bool,
+) -> T:
+    """Run a hybrid query, retrying on the primary key if the table refuses
+    `_rowid`.
+
+    `_rowid` is asked for first rather than paying a round trip up front to
+    find out: only a MemWAL table refuses, and the refusal has to be handled
+    anyway -- a write spec installed elsewhere can arrive between any two
+    queries. It costs nothing server-side: it is raised before the query is
+    planned. `fused` reports whether the attempt already used the key.
+    """
+    try:
+        return run()
+    except Exception as e:
+        _check_refusal(
+            e, fused=fused(), caller_requested_row_id=caller_requested_row_id
+        )
+        # Learned, so this retry and later queries go straight to the key.
+        table._note_hybrid_pk_fusion()
+        return run()
+
+
+async def with_pk_fallback_async(
+    run: Callable[[], Awaitable[T]],
+    table,
+    *,
+    fused: Callable[[], bool],
+    caller_requested_row_id: bool,
+) -> T:
+    """The async form of `with_pk_fallback`."""
+    try:
+        return await run()
+    except Exception as e:
+        _check_refusal(
+            e, fused=fused(), caller_requested_row_id=caller_requested_row_id
+        )
+        table._note_hybrid_pk_fusion()
+        return await run()
diff --git a/python/python/lancedb/query.py b/python/python/lancedb/query.py
index cef89e63c..c42344ad3 100644
--- a/python/python/lancedb/query.py
+++ b/python/python/lancedb/query.py
@@ -41,6 +41,7 @@ from .rerankers.rrf import RRFReranker
 from .rerankers.util import check_reranker_result
 from .schema import is_blob_like_field, schema_has_blob_field
 from .util import flatten_columns
+from . import _wal_hybrid  # WAL-PK-FUSION: delete.
 from ._blob import (
     BLOB_MODE_TO_HANDLING,
     FetchBlobsAsync,
@@ -2207,6 +2208,12 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
         self._phrase_query = None
         self._lower_bound = None
         self._upper_bound = None
+        # WAL-PK-FUSION: delete both, and the `with_row_id` override below.
+        # Set by `_create_query_builders` when the legs are fused on the key.
+        self._pk_fusion: Optional[_wal_hybrid.PkFusion] = None
+        # `_with_row_id` is also turned on by `rerank(return_score="all")` for
+        # the reranker's own use, so it cannot answer "did the caller ask?".
+        self._caller_requested_row_id = False
 
     def _validate_query(self, query, vector=None, text=None):
         if query is not None and (vector is not None or text is not None):
@@ -2228,6 +2235,16 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
 
         return vector_query, text_query
 
+    def with_row_id(self, with_row_id: bool) -> Self:  # WAL-PK-FUSION: delete.
+        """Set whether to return row ids.
+
+        Recorded separately from `_with_row_id`, which `rerank(return_score=
+        "all")` also sets for its own use — on a MemWAL table only a caller who
+        asked is refused, and the reranker still falls back to the primary key.
+        """
+        self._caller_requested_row_id = with_row_id
+        return super().with_row_id(with_row_id)
+
     def phrase_query(self, phrase_query: bool = True) -> LanceHybridQueryBuilder:
         """Set whether to use phrase query.
 
@@ -2248,16 +2265,30 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
         raise NotImplementedError("to_query_object not yet supported on a hybrid query")
 
     def to_arrow(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
+        # WAL-PK-FUSION: without the fallback, this body is `_run_hybrid`'s.
+        return _wal_hybrid.with_pk_fallback(
+            lambda: self._run_hybrid(timeout=timeout),
+            self._table,
+            fused=lambda: self._pk_fusion is not None,
+            caller_requested_row_id=self._caller_requested_row_id,
+        )
+
+    def _run_hybrid(self, *, timeout: Optional[timedelta] = None) -> pa.Table:
         self._create_query_builders()
+        fts_query, vector_query = self._fts_query, self._vector_query
+        # WAL-PK-FUSION: without the fallback, both legs always ask for row ids.
+        if self._pk_fusion is None:
+            fts_query = fts_query.with_row_id(True)
+            vector_query = vector_query.with_row_id(True)
         with ThreadPoolExecutor() as executor:
-            fts_future = executor.submit(
-                self._fts_query.with_row_id(True).to_arrow, timeout=timeout
-            )
-            vector_future = executor.submit(
-                self._vector_query.with_row_id(True).to_arrow, timeout=timeout
-            )
+            fts_future = executor.submit(fts_query.to_arrow, timeout=timeout)
+            vector_future = executor.submit(vector_query.to_arrow, timeout=timeout)
             fts_results = fts_future.result()
             vector_results = vector_future.result()
+        if self._pk_fusion is not None:  # WAL-PK-FUSION: delete.
+            vector_results, fts_results = self._pk_fusion.stamp(
+                vector_results, fts_results
+            )
 
         results = self._combine_hybrid_results(
             fts_results=fts_results,
@@ -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))
diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py
index 2a6512fc5..44d16211c 100644
--- a/python/python/lancedb/remote/table.py
+++ b/python/python/lancedb/remote/table.py
@@ -1036,6 +1036,14 @@ class RemoteTable(Table):
         """Read the installed LsmWriteSpec, or ``None``."""
         return LOOP.run(self._table.get_lsm_write_spec())
 
+    # WAL-PK-FUSION: delete both hooks.
+    def _hybrid_pk_fusion_learned(self) -> bool:
+        # `self._table` is the async table, which owns the Rust handle.
+        return self._table._hybrid_pk_fusion_learned()
+
+    def _note_hybrid_pk_fusion(self) -> None:
+        self._table._note_hybrid_pk_fusion()
+
     def checkpoint_lsm(self) -> None:
         """Synchronous version of
         [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py
index d87012bcf..0dc927b29 100644
--- a/python/python/lancedb/table.py
+++ b/python/python/lancedb/table.py
@@ -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.
 
diff --git a/python/python/tests/test_wal_hybrid.py b/python/python/tests/test_wal_hybrid.py
new file mode 100644
index 000000000..11a9755e0
--- /dev/null
+++ b/python/python/tests/test_wal_hybrid.py
@@ -0,0 +1,219 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""WAL-PK-FUSION: delete this file with `lancedb/_wal_hybrid.py`."""
+
+from unittest import mock
+
+import lancedb
+import pyarrow as pa
+import pytest
+from lancedb._wal_hybrid import WAL_ROW_ID_UNSUPPORTED, with_pk_fallback
+from lancedb.query import LanceHybridQueryBuilder
+from lancedb.rerankers import RRFReranker
+
+
+def _wal_table(schema: pa.Schema) -> mock.Mock:
+    """A MemWAL-backed table a previous query has already been refused
+    `_rowid` on, so the fusion goes straight to the primary key."""
+    table = mock.Mock()
+    table._hybrid_pk_fusion_learned.return_value = True
+    table.schema = schema
+    return table
+
+
+_PK_SCHEMA = pa.schema(
+    [
+        pa.field(
+            "id",
+            pa.string(),
+            metadata={b"lance-schema:unenforced-primary-key:position": b"1"},
+        ),
+        pa.field("text", pa.string()),
+    ]
+)
+
+
+def _wal_hybrid(**select):
+    builder = (
+        LanceHybridQueryBuilder(_wal_table(_PK_SCHEMA)).vector([0.1, 0.2]).text("puppy")
+    )
+    if "columns" in select:
+        builder = builder.select(select["columns"])
+    return builder
+
+
+def test_wal_hybrid_projects_the_key_instead_of_asking_for_row_ids():
+    builder = _wal_hybrid(columns=["text"])
+    builder._create_query_builders()
+
+    assert builder._pk_fusion.pk_columns == ["id"]
+    assert builder._pk_fusion.injected == ["id"]
+    for leg in (builder._vector_query, builder._fts_query):
+        assert leg._columns == ["text", "id"]
+        # A MemWAL table rejects with_row_id before it plans.
+        assert leg._with_row_id is not True
+
+
+def test_wal_hybrid_selecting_the_key_injects_nothing():
+    builder = _wal_hybrid(columns=["text", "id"])
+    builder._create_query_builders()
+    assert builder._pk_fusion.injected == []
+
+
+def test_wal_hybrid_without_a_projection_leaves_it_alone():
+    builder = _wal_hybrid()
+    builder._create_query_builders()
+    # Select-all already returns the key.
+    assert builder._pk_fusion.injected == []
+    assert builder._vector_query._columns is None
+
+
+def test_base_table_hybrid_still_joins_on_row_ids():
+    table = mock.Mock()
+    table._hybrid_pk_fusion_learned.return_value = False
+    builder = LanceHybridQueryBuilder(table).vector([0.1, 0.2]).text("puppy")
+    builder._create_query_builders()
+
+    # `to_arrow` is what turns row ids on for this path; all that matters here
+    # is that no key was resolved and no projection was touched.
+    assert builder._pk_fusion is None
+    assert builder._vector_query._columns is None
+
+
+def test_wal_hybrid_with_use_lsm_false_joins_on_row_ids():
+    builder = _wal_hybrid().use_lsm(False)
+    builder._create_query_builders()
+    assert builder._pk_fusion is None
+
+
+def test_wal_hybrid_refuses_an_explicit_with_row_id():
+    builder = _wal_hybrid().with_row_id(True)
+    with pytest.raises(NotImplementedError, match="use_lsm"):
+        builder._create_query_builders()
+
+
+def test_wal_hybrid_return_score_all_is_not_refused():
+    """`return_score="all"` turns row ids on for the reranker, not the caller."""
+    builder = _wal_hybrid().rerank(RRFReranker(return_score="all"))
+    builder._create_query_builders()
+    assert builder._pk_fusion.pk_columns == ["id"]
+
+
+def test_wal_hybrid_fuses_on_the_key_and_drops_it_again():
+    builder = _wal_hybrid(columns=["text"])
+    builder._create_query_builders()
+
+    vector_results = pa.table(
+        {"text": ["a", "b"], "id": ["a", "b"], "_distance": [0.1, 0.2]}
+    )
+    fts_results = pa.table({"text": ["b", "c"], "id": ["b", "c"], "_score": [1.0, 2.0]})
+    vector_results, fts_results = builder._pk_fusion.stamp(vector_results, fts_results)
+
+    combined = LanceHybridQueryBuilder._combine_hybrid_results(
+        fts_results=fts_results,
+        vector_results=vector_results,
+        norm="score",
+        fts_query="puppy",
+        reranker=RRFReranker(),
+        limit=10,
+        with_row_ids=True,
+    )
+    results = builder._finish_hybrid_results(combined)
+
+    assert "_rowid" not in results.column_names, "the surrogate is internal"
+    assert "id" not in results.column_names, "the caller selected only `text`"
+    assert sorted(results.column("text").to_pylist()) == ["a", "b", "c"]
+
+
+@pytest.mark.parametrize(
+    "row_id_first", [True, False], ids=["row_id_first", "rerank_first"]
+)
+def test_wal_hybrid_refuses_row_id_even_with_return_score_all(row_id_first: bool):
+    """`return_score="all"` sets the same flag the caller's request does, so
+    asking for both must still be refused — in either call order."""
+    builder = _wal_hybrid()
+    reranker = RRFReranker(return_score="all")
+    if row_id_first:
+        builder = builder.with_row_id(True).rerank(reranker)
+    else:
+        builder = builder.rerank(reranker).with_row_id(True)
+
+    with pytest.raises(NotImplementedError, match="use_lsm"):
+        builder._create_query_builders()
+
+
+def test_wal_hybrid_refuses_a_blob_projection_with_return_score_all():
+    """A projected blob needs a real row id to fetch through, which the
+    reranker turning `_with_row_id` on does not provide."""
+    schema = pa.schema(
+        [
+            pa.field(
+                "id",
+                pa.string(),
+                metadata={b"lance-schema:unenforced-primary-key:position": b"1"},
+            ),
+            pa.field("text", pa.string()),
+            lancedb.blob("blob"),
+        ]
+    )
+    builder = (
+        LanceHybridQueryBuilder(_wal_table(schema))
+        .vector([0.1, 0.2])
+        .text("puppy")
+        .select(["blob"])
+        .rerank(RRFReranker(return_score="all"))
+    )
+
+    with pytest.raises(NotImplementedError, match="blob"):
+        builder._create_query_builders()
+
+
+_REFUSAL = RuntimeError(
+    "Bad request: the MemWAL LSM scanner does not support with_row_id"
+)
+
+
+def test_fallback_retries_on_the_key_and_remembers_it():
+    table = mock.Mock()
+    attempts = []
+
+    def run():
+        attempts.append(len(attempts))
+        if len(attempts) == 1:
+            raise _REFUSAL
+        return "fused"
+
+    result = with_pk_fallback(
+        run, table, fused=lambda: False, caller_requested_row_id=False
+    )
+
+    assert result == "fused"
+    assert len(attempts) == 2
+    table._note_hybrid_pk_fusion.assert_called_once()
+
+
+def test_fallback_explains_the_refusal_to_a_caller_who_asked_for_row_ids():
+    def run():
+        raise _REFUSAL
+
+    with pytest.raises(NotImplementedError, match="use_lsm") as info:
+        with_pk_fallback(
+            run, mock.Mock(), fused=lambda: False, caller_requested_row_id=True
+        )
+    assert str(info.value) == WAL_ROW_ID_UNSUPPORTED
+
+
+@pytest.mark.parametrize(
+    "error, fused",
+    [(ValueError("unrelated"), False), (_REFUSAL, True)],
+    ids=["other_error", "already_fused"],
+)
+def test_fallback_reraises_what_it_cannot_retry(error, fused):
+    def run():
+        raise error
+
+    table = mock.Mock()
+    with pytest.raises(type(error)):
+        with_pk_fallback(run, table, fused=lambda: fused, caller_requested_row_id=False)
+    table._note_hybrid_pk_fusion.assert_not_called()
diff --git a/python/src/table.rs b/python/src/table.rs
index bf5250732..e7b6f1c6f 100644
--- a/python/src/table.rs
+++ b/python/src/table.rs
@@ -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
diff --git a/rust/lancedb/src/query.rs b/rust/lancedb/src/query.rs
index ab1bdfc4a..c8b19f013 100644
--- a/rust/lancedb/src/query.rs
+++ b/rust/lancedb/src/query.rs
@@ -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))
diff --git a/rust/lancedb/src/query/wal_fusion.rs b/rust/lancedb/src/query/wal_fusion.rs
new file mode 100644
index 000000000..b438f6150
--- /dev/null
+++ b/rust/lancedb/src/query/wal_fusion.rs
@@ -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}");
+    }
+}
diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs
index 6bbb49adf..d6abfea9a 100644
--- a/rust/lancedb/src/remote/table.rs
+++ b/rust/lancedb/src/remote/table.rs
@@ -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"
+        );
+    }
 }
diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs
index fbec73dfb..dcb3e1578 100644
--- a/rust/lancedb/src/table.rs
+++ b/rust/lancedb/src/table.rs
@@ -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`.

From d0fb43b0b45c959d44655f1471b469890787054f Mon Sep 17 00:00:00 2001
From: LanceDB Robot <robot@lancedb.com>
Date: Fri, 25 Sep 2026 07:02:00 -0700
Subject: [PATCH 2/2] chore: update lance dependency to v13.0.0-beta.14 (#4324)

Update the Rust workspace Lance dependencies and Java lance-core from
v13.0.0-beta.12 to
[v13.0.0-beta.14](https://github.com/lance-format/lance/releases/tag/v13.0.0-beta.14),
including the refreshed Cargo lockfile.

No compatibility fixes were required; `cargo clippy --quiet --workspace
--tests --all-features -- -D warnings`, `cargo fmt --all --quiet`, and
`git diff --check` passed.
---
 Cargo.lock   | 88 ++++++++++++++++++++++++++--------------------------
 Cargo.toml   | 28 ++++++++---------
 java/pom.xml |  2 +-
 3 files changed, 59 insertions(+), 59 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 367ba7805..ed0fad6f3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -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,8 +5371,8 @@ dependencies = [
 
 [[package]]
 name = "lance-encoding"
-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",
@@ -5405,8 +5405,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",
@@ -5438,8 +5438,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",
@@ -5453,8 +5453,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",
@@ -5524,8 +5524,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",
@@ -5547,8 +5547,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",
@@ -5590,8 +5590,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",
@@ -5605,8 +5605,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",
@@ -5620,8 +5620,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",
@@ -5674,8 +5674,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",
@@ -5689,8 +5689,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",
@@ -5730,8 +5730,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",
@@ -5744,8 +5744,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",
diff --git a/Cargo.toml b/Cargo.toml
index c64305048..15770bebc 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,20 +13,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 = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-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 = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+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
diff --git a/java/pom.xml b/java/pom.xml
index 7fe65841f..752dacefd 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -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>