diff --git a/Cargo.lock b/Cargo.lock
index 34d0a5f5f..bde776408 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,7 +5371,7 @@ dependencies = [
 
 [[package]]
 name = "lance-encoding"
-version = "13.0.0-beta.12"
+version = "13.0.0-beta.14"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5404,8 +5404,8 @@ dependencies = [
 
 [[package]]
 name = "lance-file"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5437,8 +5437,8 @@ dependencies = [
 
 [[package]]
 name = "lance-geo"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "datafusion",
  "geo-traits",
@@ -5452,8 +5452,8 @@ dependencies = [
 
 [[package]]
 name = "lance-index"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -5523,8 +5523,8 @@ dependencies = [
 
 [[package]]
 name = "lance-index-core"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5546,8 +5546,8 @@ dependencies = [
 
 [[package]]
 name = "lance-io"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5589,8 +5589,8 @@ dependencies = [
 
 [[package]]
 name = "lance-linalg"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5604,8 +5604,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow",
  "async-trait",
@@ -5619,8 +5619,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace-impls"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow",
  "arrow-ipc",
@@ -5673,8 +5673,8 @@ dependencies = [
 
 [[package]]
 name = "lance-select"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5688,8 +5688,8 @@ dependencies = [
 
 [[package]]
 name = "lance-table"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5729,8 +5729,8 @@ dependencies = [
 
 [[package]]
 name = "lance-testing"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -5743,8 +5743,8 @@ dependencies = [
 
 [[package]]
 name = "lance-tokenizer"
-version = "13.0.0-beta.12"
-source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.12#5d2946fa452c61cb018144f3200892b6dd4a8b4f"
+version = "13.0.0-beta.14"
+source = "git+https://github.com/lance-format/lance.git?tag=v13.0.0-beta.14#b80ac1d9f3c24b47bd8e297d03053ea35d216a9f"
 dependencies = [
  "frostem",
  "icu_segmenter",
diff --git a/Cargo.toml b/Cargo.toml
index 1779bfd74..9b85c689c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -14,20 +14,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"
 
 [workspace.dependencies]
-lance = { "version" = "=13.0.0-beta.12", default-features = false, "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=13.0.0-beta.12", default-features = false, "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=13.0.0-beta.12", default-features = false, "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { path = "vendor/lance-encoding", version = "=13.0.0-beta.12" }
-lance-arrow = { "version" = "=13.0.0-beta.12", "tag" = "v13.0.0-beta.12", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=13.0.0-beta.14", default-features = false, "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=13.0.0-beta.14", default-features = false, "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=13.0.0-beta.14", default-features = false, "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { path = "vendor/lance-encoding", version = "=13.0.0-beta.14" }
+lance-arrow = { "version" = "=13.0.0-beta.14", "tag" = "v13.0.0-beta.14", "git" = "https://github.com/lance-format/lance.git" }
 lancedb = { path = "rust/lancedb", default-features = false }
 ahash = "0.8"
 # Note that this one does not include pyarrow
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>
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`.
diff --git a/vendor/lance-encoding/Cargo.toml b/vendor/lance-encoding/Cargo.toml
index 3819fa3d2..f8db41c60 100644
--- a/vendor/lance-encoding/Cargo.toml
+++ b/vendor/lance-encoding/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "lance-encoding"
-version = "13.0.0-beta.12"
+version = "13.0.0-beta.14"
 edition = "2024"
 license = "Apache-2.0"
 repository = "https://github.com/lance-format/lance"
@@ -13,10 +13,10 @@ rust-version = "1.91.0"
 [workspace]
 
 [dependencies]
-lance-arrow = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12" }
-lance-core = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12" }
-lance-bitpacking = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12", optional = true }
-fsst = { version = "=13.0.0-beta.12", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.12" }
+lance-arrow = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14" }
+lance-core = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14" }
+lance-bitpacking = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14", optional = true }
+fsst = { version = "=13.0.0-beta.14", git = "https://github.com/lance-format/lance.git", tag = "v13.0.0-beta.14" }
 arrow-arith = "58.0.0"
 arrow-array = "58.0.0"
 arrow-buffer = "58.0.0"
diff --git a/vendor/lance-encoding/PATCH.md b/vendor/lance-encoding/PATCH.md
index 5b5671b18..86c6d3974 100644
--- a/vendor/lance-encoding/PATCH.md
+++ b/vendor/lance-encoding/PATCH.md
@@ -1,5 +1,5 @@
 This is `rust/lance-encoding` from
-[`lance-format/lance` v13.0.0-beta.12](https://github.com/lance-format/lance/tree/5d2946fa452c61cb018144f3200892b6dd4a8b4f/rust/lance-encoding),
+[`lance-format/lance` v13.0.0-beta.14](https://github.com/lance-format/lance/tree/b80ac1d9f3c24b47bd8e297d03053ea35d216a9f/rust/lance-encoding),
 with the crate's build protos copied into `protos/`. Its manifest uses explicit
 dependencies so it can be built outside the Lance workspace.