chore: use streamed-copy lance branch

fix(remote): centralize timeline consistency (#4053 )
Centralizes remote table freshness fencing and response-version tracking in the default transport path. Covers schema and blob bypass paths, keeps explicit time-travel and cross-timeline operations unfenced, and advances freshness after refresh and index job completion.
2026-09-04 20:48:50 +00:00 · 2026-08-25 22:46:47 -07:00 · 2026-08-26 12:54:36 +08:00 · 2026-08-25 21:56:26 -05:00 · 2026-08-26 10:08:48 +08:00
16 changed files with 1966 additions and 742 deletions
@@ -1756,7 +1756,7 @@ version = "3.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -3034,7 +3034,7 @@ dependencies = [
 "libc",
 "option-ext",
 "redox_users",
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -3257,7 +3257,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
 "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -3456,7 +3456,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 [[package]]
 name = "fsst"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4561,7 +4561,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
 dependencies = [
 "hermit-abi",
 "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -4816,7 +4816,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
 [[package]]
 name = "lance"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4889,7 +4889,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4911,7 +4911,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4925,7 +4925,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4935,7 +4935,7 @@ dependencies = [
 [[package]]
 name = "lance-bitpacking"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4946,7 +4946,7 @@ dependencies = [
 [[package]]
 name = "lance-core"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4984,7 +4984,7 @@ dependencies = [
 [[package]]
 name = "lance-datafusion"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5014,7 +5014,7 @@ dependencies = [
 [[package]]
 name = "lance-datagen"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5032,7 +5032,7 @@ dependencies = [
 [[package]]
 name = "lance-derive"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5042,7 +5042,7 @@ dependencies = [
 [[package]]
 name = "lance-encoding"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5076,7 +5076,7 @@ dependencies = [
 [[package]]
 name = "lance-file"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5108,7 +5108,7 @@ dependencies = [
 [[package]]
 name = "lance-index"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5173,7 +5173,7 @@ dependencies = [
 [[package]]
 name = "lance-index-core"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5196,7 +5196,7 @@ dependencies = [
 [[package]]
 name = "lance-io"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5237,7 +5237,7 @@ dependencies = [
 [[package]]
 name = "lance-linalg"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5252,7 +5252,7 @@ dependencies = [
 [[package]]
 name = "lance-namespace"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5265,7 +5265,7 @@ dependencies = [
 [[package]]
 name = "lance-namespace-impls"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5319,7 +5319,7 @@ dependencies = [
 [[package]]
 name = "lance-select"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5334,7 +5334,7 @@ dependencies = [
 [[package]]
 name = "lance-table"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5375,7 +5375,7 @@ dependencies = [
 [[package]]
 name = "lance-testing"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5389,7 +5389,7 @@ dependencies = [
 [[package]]
 name = "lance-tokenizer"
 version = "12.0.0-beta.2"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
+source = "git+https://github.com/jackye1995/lance.git?branch=jack%2Favoid-object-store-copy-for-fts-remap#318e7d0ae71a9945b91df86ede8f23bdb060b4ce"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -6240,7 +6240,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -7632,8 +7632,8 @@ version = "0.14.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7"
 dependencies = [
- "heck 0.5.0",
- "itertools 0.14.0",
+ "heck 0.4.1",
+ "itertools 0.11.0",
 "log",
 "multimap",
 "petgraph",
@@ -7652,7 +7652,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b"
 dependencies = [
 "anyhow",
- "itertools 0.14.0",
+ "itertools 0.11.0",
 "proc-macro2",
 "quote",
 "syn 2.0.117",
@@ -7923,7 +7923,7 @@ dependencies = [
 "once_cell",
 "socket2 0.6.3",
 "tracing",
- "windows-sys 0.60.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -8701,7 +8701,7 @@ dependencies = [
 "errno",
 "libc",
 "linux-raw-sys",
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -8772,7 +8772,7 @@ dependencies = [
 "security-framework",
 "security-framework-sys",
 "webpki-root-certs",
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -9328,7 +9328,7 @@ version = "0.8.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c1c97747dbf44bb1ca44a561ece23508e99cb592e862f22222dcf42f51d1e451"
 dependencies = [
- "heck 0.5.0",
+ "heck 0.4.1",
 "proc-macro2",
 "quote",
 "syn 2.0.117",
@@ -9340,7 +9340,7 @@ version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "54254b8531cafa275c5e096f62d48c81435d1015405a91198ddb11e967301d40"
 dependencies = [
- "heck 0.5.0",
+ "heck 0.4.1",
 "proc-macro2",
 "quote",
 "syn 2.0.117",
@@ -9762,7 +9762,7 @@ dependencies = [
 "getrandom 0.4.2",
 "once_cell",
 "rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -10739,7 +10739,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.59.0",
 ]

 [[package]]
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=12.0.0-beta.2", default-features = false, "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=12.0.0-beta.2", "tag" = "v12.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=12.0.0-beta.2", default-features = false, "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-core = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-datagen = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-file = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-io = { "version" = "=12.0.0-beta.2", default-features = false, "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-index = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-linalg = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-namespace = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-namespace-impls = { "version" = "=12.0.0-beta.2", default-features = false, "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-table = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-testing = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-datafusion = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-encoding = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
+lance-arrow = { "version" = "=12.0.0-beta.2", "branch" = "jack/avoid-object-store-copy-for-fts-remap", "git" = "https://github.com/jackye1995/lance.git" }
 lancedb = { path = "rust/lancedb", default-features = false }
 ahash = "0.8"
 # Note that this one does not include pyarrow
@@ -2585,7 +2585,24 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      );
    });

-    test("full text search if no embedding function provided", async () => {
+    test("full text search if only an unrelated embedding function is registered", async () => {
+      register("unused")(
+        class extends EmbeddingFunction<string> {
+          ndims() {
+            return 3;
+          }
+          embeddingDataType() {
+            return new Float32();
+          }
+          async computeQueryEmbeddings(_data: string) {
+            return [1, 2, 3];
+          }
+          async computeSourceEmbeddings(data: string[]) {
+            return data.map(() => [1, 2, 3]);
+          }
+        },
+      );
+
      const db = await connect(tmpDir.name);
      const data = [
        { text: "hello world", vector: [0.1, 0.2, 0.3] },
@@ -2607,6 +2624,306 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      expect(results2[0].text).toBe(data[1].text);
    });

+    test("auto search stays consistent with the active revision", async () => {
+      let initCalls = 0;
+      let queryCalls = 0;
+      let markStarted!: () => void;
+      const started = new Promise<void>((resolve) => {
+        markStarted = resolve;
+      });
+      let releaseEmbedding!: () => void;
+      const embeddingReleased = new Promise<void>((resolve) => {
+        releaseEmbedding = resolve;
+      });
+
+      @register("refresh-test")
+      class TestEmbedding extends EmbeddingFunction<string> {
+        async init() {
+          initCalls += 1;
+        }
+        ndims() {
+          return 1;
+        }
+        embeddingDataType() {
+          return new arrow.Float32();
+        }
+        async computeQueryEmbeddings(value: string) {
+          queryCalls += 1;
+          if (value === "blocked") {
+            markStarted();
+            await embeddingReleased;
+          }
+          return value === "greetings" ? [0.1] : [0.2];
+        }
+        async computeSourceEmbeddings(values: string[]) {
+          return values.map((value) =>
+            value === "hello world" ? [0.1] : [0.2],
+          );
+        }
+      }
+
+      const writer = await connect(tmpDir.name);
+      await writer.createTable("test", [{ text: "plain", vector: [0.0] }]);
+      const reader = await connect(tmpDir.name, {
+        readConsistencyInterval: 0,
+      });
+      const tracked = await reader.openTable("test");
+      type SnapshotCountingNative = {
+        querySnapshot: () => Promise<unknown>;
+      };
+      const native = (tracked as unknown as { inner: SnapshotCountingNative })
+        .inner;
+      const querySnapshot = native.querySnapshot.bind(native);
+      let snapshotCalls = 0;
+      native.querySnapshot = async () => {
+        snapshotCalls += 1;
+        return await querySnapshot();
+      };
+      const autoQuery = tracked.search("greetings").select(["text"]).limit(1);
+
+      const func = new TestEmbedding();
+      const schema = LanceSchema({
+        text: func.sourceField(new arrow.Utf8()),
+        vector: func.vectorField(),
+      });
+      const data = [{ text: "hello world" }, { text: "goodbye world" }];
+      await writer.createTable("test", data, { mode: "overwrite", schema });
+      const baselineInitCalls = initCalls;
+
+      expect(
+        (await tracked.schema()).metadata.get("embedding_functions"),
+      ).toBeDefined();
+      const results = await autoQuery.toArray();
+      expect(results[0].text).toBe(data[0].text);
+      expect(initCalls).toBe(baselineInitCalls + 1);
+      expect(queryCalls).toBe(1);
+      expect(snapshotCalls).toBe(1);
+
+      const repeatedResults = await autoQuery.toArray();
+      expect(repeatedResults[0].text).toBe(data[0].text);
+      expect(initCalls).toBe(baselineInitCalls + 1);
+      expect(queryCalls).toBe(1);
+      expect(snapshotCalls).toBe(2);
+
+      const pending = tracked
+        .search("blocked")
+        .select(["text"])
+        .limit(1)
+        .toArray();
+      await started;
+
+      const ftsData = [
+        { text: "greetings from full text", vector: [0.0] },
+        { text: "blocked from full text", vector: [0.0] },
+      ];
+      const ftsTable = await writer.createTable("test", ftsData, {
+        mode: "overwrite",
+      });
+      await ftsTable.createIndex("text", { config: Index.fts() });
+      releaseEmbedding();
+
+      const pendingResults = await pending;
+      expect(pendingResults[0].text).toBe(data[1].text);
+
+      expect(
+        (await tracked.schema()).metadata.get("embedding_functions"),
+      ).toBeUndefined();
+      const ftsResults = await autoQuery.toArray();
+      expect(ftsResults[0].text).toBe(ftsData[0].text);
+    });
+
+    test("auto search keeps newer preparation during a revision race", async () => {
+      let aCalls = 0;
+      let bCalls = 0;
+      let markAStarted!: () => void;
+      const aStarted = new Promise<void>((resolve) => {
+        markAStarted = resolve;
+      });
+      let releaseA!: () => void;
+      const aReleased = new Promise<void>((resolve) => {
+        releaseA = resolve;
+      });
+      let markBStarted!: () => void;
+      const bStarted = new Promise<void>((resolve) => {
+        markBStarted = resolve;
+      });
+      let releaseB!: () => void;
+      const bReleased = new Promise<void>((resolve) => {
+        releaseB = resolve;
+      });
+
+      @register("race-a")
+      class EmbeddingA extends EmbeddingFunction<string> {
+        ndims() {
+          return 1;
+        }
+        embeddingDataType() {
+          return new arrow.Float32();
+        }
+        async computeQueryEmbeddings() {
+          aCalls += 1;
+          markAStarted();
+          await aReleased;
+          return [0.1];
+        }
+        async computeSourceEmbeddings(values: string[]) {
+          return values.map(() => [0.1]);
+        }
+      }
+
+      @register("race-b")
+      class EmbeddingB extends EmbeddingFunction<string> {
+        ndims() {
+          return 1;
+        }
+        embeddingDataType() {
+          return new arrow.Float32();
+        }
+        async computeQueryEmbeddings() {
+          bCalls += 1;
+          markBStarted();
+          await bReleased;
+          return [0.2];
+        }
+        async computeSourceEmbeddings(values: string[]) {
+          return values.map(() => [0.2]);
+        }
+      }
+
+      const writer = await connect(tmpDir.name);
+      const embeddingA = new EmbeddingA();
+      const schemaA = LanceSchema({
+        text: embeddingA.sourceField(new arrow.Utf8()),
+        vector: embeddingA.vectorField(),
+      });
+      await writer.createTable("race", [{ text: "revision a" }], {
+        schema: schemaA,
+      });
+      const reader = await connect(tmpDir.name, {
+        readConsistencyInterval: 0,
+      });
+      const tracked = await reader.openTable("race");
+      const query = tracked.search("query");
+
+      const first = query.toArray();
+      await aStarted;
+
+      const embeddingB = new EmbeddingB();
+      const schemaB = LanceSchema({
+        text: embeddingB.sourceField(new arrow.Utf8()),
+        vector: embeddingB.vectorField(),
+      });
+      await writer.createTable("race", [{ text: "revision b" }], {
+        mode: "overwrite",
+        schema: schemaB,
+      });
+      const second = query.toArray();
+      await bStarted;
+
+      releaseA();
+      releaseB();
+      await Promise.all([first, second]);
+      expect(aCalls).toBe(1);
+      expect(bCalls).toBe(1);
+    });
+
+    test("stale FTS routing keeps newer vector preparation", async () => {
+      let vectorCalls = 0;
+      let markVectorStarted!: () => void;
+      const vectorStarted = new Promise<void>((resolve) => {
+        markVectorStarted = resolve;
+      });
+      let releaseVector!: () => void;
+      const vectorReleased = new Promise<void>((resolve) => {
+        releaseVector = resolve;
+      });
+
+      @register("stale-fts-race")
+      class RaceEmbedding extends EmbeddingFunction<string> {
+        ndims() {
+          return 1;
+        }
+        embeddingDataType() {
+          return new arrow.Float32();
+        }
+        async computeQueryEmbeddings() {
+          vectorCalls += 1;
+          markVectorStarted();
+          await vectorReleased;
+          return [0.1];
+        }
+        async computeSourceEmbeddings(values: string[]) {
+          return values.map(() => [0.1]);
+        }
+      }
+
+      const writer = await connect(tmpDir.name);
+      const ftsTable = await writer.createTable("stale_fts", [
+        { text: "hello", vector: [0.0] },
+      ]);
+      await ftsTable.createIndex("text", { config: Index.fts() });
+
+      const reader = await connect(tmpDir.name, {
+        readConsistencyInterval: 0,
+      });
+      const tracked = await reader.openTable("stale_fts");
+      type Snapshot = {
+        schema: () => Promise<Buffer>;
+      };
+      type NativeWithSnapshot = {
+        querySnapshot: () => Promise<Snapshot>;
+      };
+      const native = (tracked as unknown as { inner: NativeWithSnapshot })
+        .inner;
+      const querySnapshot = native.querySnapshot.bind(native);
+      let snapshotCalls = 0;
+      let markStaleSchemaStarted!: () => void;
+      const staleSchemaStarted = new Promise<void>((resolve) => {
+        markStaleSchemaStarted = resolve;
+      });
+      let releaseStaleSchema!: () => void;
+      const staleSchemaReleased = new Promise<void>((resolve) => {
+        releaseStaleSchema = resolve;
+      });
+      native.querySnapshot = async () => {
+        const snapshot = await querySnapshot();
+        snapshotCalls += 1;
+        if (snapshotCalls === 1) {
+          const schema = snapshot.schema.bind(snapshot);
+          snapshot.schema = async () => {
+            markStaleSchemaStarted();
+            await staleSchemaReleased;
+            return await schema();
+          };
+        }
+        return snapshot;
+      };
+
+      const query = tracked.search("hello");
+      const staleFtsExecution = query.toArray();
+      await staleSchemaStarted;
+
+      const embedding = new RaceEmbedding();
+      const vectorSchema = LanceSchema({
+        text: embedding.sourceField(new arrow.Utf8()),
+        vector: embedding.vectorField(),
+      });
+      await writer.createTable("stale_fts", [{ text: "hello" }], {
+        mode: "overwrite",
+        schema: vectorSchema,
+      });
+
+      const vectorExecution = query.toArray();
+      await vectorStarted;
+      releaseStaleSchema();
+      await staleFtsExecution;
+      releaseVector();
+      await vectorExecution;
+
+      await query.toArray();
+      expect(vectorCalls).toBe(1);
+    });
+
    test("tokenizes FTS queries by column or index name", async () => {
      const db = await connect(tmpDir.name);
      const data = [
@@ -3157,6 +3474,30 @@ describe("column name options", () => {
    expect(results[1].query_index).toBe(1);
  });

+  test("observes promised additional vectors while the query is pending", async () => {
+    const initialVector = new Promise<number[]>(() => undefined);
+    const query = table.query().nearestTo(initialVector);
+    const unhandled: unknown[] = [];
+    const onUnhandled = (reason: unknown) => unhandled.push(reason);
+    process.on("unhandledRejection", onUnhandled);
+
+    try {
+      query.addQueryVector(Promise.reject(new Error("extra vector failed")));
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(unhandled).toEqual([]);
+
+      const rejectedQuery = table
+        .query()
+        .nearestTo([0.1, 0.2])
+        .addQueryVector(Promise.reject(new Error("consumed vector failed")));
+      await expect(rejectedQuery.toArray()).rejects.toThrow(
+        "consumed vector failed",
+      );
+    } finally {
+      process.off("unhandledRejection", onUnhandled);
+    }
+  });
+
  test("index and search multivectors", async () => {
    const db = await connect(tmpDir.name);
    const data = [];
@@ -100,6 +100,29 @@ export interface FullTextSearchOptions {
  columns?: string | string[];
 }

+function nearestToNative(
+  inner: NativeQuery,
+  vector: Awaited<IntoVector>,
+): NativeVectorQuery {
+  const raw = Array.isArray(vector) ? null : extractVectorBuffer(vector);
+  if (raw) {
+    return inner.nearestToRaw(raw.data, raw.dtype);
+  }
+  return inner.nearestTo(Float32Array.from(vector as number[]));
+}
+
+function addQueryVectorToNative(
+  inner: NativeVectorQuery,
+  vector: Awaited<IntoVector>,
+) {
+  const raw = Array.isArray(vector) ? null : extractVectorBuffer(vector);
+  if (raw) {
+    inner.addQueryVectorRaw(raw.data, raw.dtype);
+  } else {
+    inner.addQueryVector(Float32Array.from(vector as number[]));
+  }
+}
+
 /** Common methods supported by all query types
 *
 * @see {@link Query}
@@ -499,6 +522,13 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
    super(inner);
  }

+  /**
+   * @hidden
+   */
+  protected doVectorCall(fn: (inner: NativeVectorQuery) => void) {
+    super.doCall(fn);
+  }
+
  /**
   * Set the number of partitions to search (probe)
   *
@@ -526,7 +556,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * the minimum and maximum to the same value.
   */
  nprobes(nprobes: number): VectorQuery {
-    super.doCall((inner) => inner.nprobes(nprobes));
+    this.doVectorCall((inner) => inner.nprobes(nprobes));

    return this;
  }
@@ -540,7 +570,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * but will also increase latency.
   */
  minimumNprobes(minimumNprobes: number): VectorQuery {
-    super.doCall((inner) => inner.minimumNprobes(minimumNprobes));
+    this.doVectorCall((inner) => inner.minimumNprobes(minimumNprobes));
    return this;
  }

@@ -554,7 +584,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * potential false negatives.
   */
  maximumNprobes(maximumNprobes: number): VectorQuery {
-    super.doCall((inner) => inner.maximumNprobes(maximumNprobes));
+    this.doVectorCall((inner) => inner.maximumNprobes(maximumNprobes));
    return this;
  }

@@ -567,7 +597,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * `undefined` means no lower or upper bound.
   */
  distanceRange(lowerBound?: number, upperBound?: number): VectorQuery {
-    super.doCall((inner) => inner.distanceRange(lowerBound, upperBound));
+    this.doVectorCall((inner) => inner.distanceRange(lowerBound, upperBound));
    return this;
  }

@@ -581,7 +611,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * also increase the latency of your query. The default value is 1.5*limit.
   */
  ef(ef: number): VectorQuery {
-    super.doCall((inner) => inner.ef(ef));
+    this.doVectorCall((inner) => inner.ef(ef));
    return this;
  }

@@ -595,7 +625,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * whose data type is a fixed-size-list of floats.
   */
  column(column: string): VectorQuery {
-    super.doCall((inner) => inner.column(column));
+    this.doVectorCall((inner) => inner.column(column));
    return this;
  }

@@ -616,7 +646,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
  distanceType(
    distanceType: Required<IvfPqOptions>["distanceType"],
  ): VectorQuery {
-    super.doCall((inner) => inner.distanceType(distanceType));
+    this.doVectorCall((inner) => inner.distanceType(distanceType));
    return this;
  }

@@ -650,7 +680,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * distance between the query vector and the actual uncompressed vector.
   */
  refineFactor(refineFactor: number): VectorQuery {
-    super.doCall((inner) => inner.refineFactor(refineFactor));
+    this.doVectorCall((inner) => inner.refineFactor(refineFactor));
    return this;
  }

@@ -675,7 +705,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * factor can often help restore some of the results lost by post filtering.
   */
  postfilter(): VectorQuery {
-    super.doCall((inner) => inner.postfilter());
+    this.doVectorCall((inner) => inner.postfilter());
    return this;
  }

@@ -689,7 +719,7 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * calculate your recall to select an appropriate value for nprobes.
   */
  bypassVectorIndex(): VectorQuery {
-    super.doCall((inner) => inner.bypassVectorIndex());
+    this.doVectorCall((inner) => inner.bypassVectorIndex());
    return this;
  }

@@ -705,35 +735,31 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   */
  addQueryVector(vector: IntoVector): VectorQuery {
    if (vector instanceof Promise) {
+      // Observe the promise as soon as it is accepted. The existing native
+      // query may still be pending, and delaying observation until it resolves
+      // can otherwise surface a fast rejection as unhandled.
+      const settledVector = vector.then(
+        (value) => ({ status: "fulfilled" as const, value }),
+        (reason) => ({ status: "rejected" as const, reason }),
+      );
      const res = (async () => {
-        try {
-          const v = await vector;
-          // biome-ignore lint/suspicious/noExplicitAny: we need to get the `inner`, but js has no package scoping
-          const value: any = this.addQueryVector(v);
-          const inner = value.inner as
-            | NativeVectorQuery
-            | Promise<NativeVectorQuery>;
-          return inner;
-        } catch (e) {
-          return Promise.reject(e);
+        const inner = await this.getInner();
+        const outcome = await settledVector;
+        if (outcome.status === "rejected") {
+          throw outcome.reason;
        }
+        addQueryVectorToNative(inner, outcome.value);
+        return inner;
      })();
      return new VectorQuery(res);
    } else {
-      super.doCall((inner) => {
-        const raw = Array.isArray(vector) ? null : extractVectorBuffer(vector);
-        if (raw) {
-          inner.addQueryVectorRaw(raw.data, raw.dtype);
-        } else {
-          inner.addQueryVector(Float32Array.from(vector as number[]));
-        }
-      });
+      this.doVectorCall((inner) => addQueryVectorToNative(inner, vector));
      return this;
    }
  }

  rerank(reranker: Reranker): VectorQuery {
-    super.doCall((inner) =>
+    this.doVectorCall((inner) =>
      inner.rerank(async (args) => {
        const vecResults = await fromBufferToRecordBatch(args.vecResults);
        const ftsResults = await fromBufferToRecordBatch(args.ftsResults);
@@ -752,6 +778,71 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
  }
 }

+/**
+ * Create a string query whose vector/FTS routing is resolved against the active
+ * table schema when the query executes.
+ *
+ * @hidden
+ */
+export function createAutoQuery(
+  table: NativeTable,
+  query: string,
+  columns: string[] | null,
+  getVector: (metadata: string) => Promise<Awaited<IntoVector>>,
+): AutoQuery {
+  type RouteSnapshot = {
+    table: NativeTable;
+    embeddingMetadata: string | undefined;
+  };
+  type CachedPreparation = {
+    metadata: string;
+    vector: Promise<Awaited<IntoVector>>;
+  };
+
+  let cachedPreparation: CachedPreparation | undefined;
+
+  const snapshotRoute = async (): Promise<RouteSnapshot> => {
+    const snapshot = await table.querySnapshot();
+    const schema = tableFromIPC(await snapshot.schema()).schema;
+    return {
+      table: snapshot,
+      embeddingMetadata: schema.metadata.get("embedding_functions"),
+    };
+  };
+
+  const createInner = async (): Promise<NativeQuery | NativeVectorQuery> => {
+    const route = await snapshotRoute();
+    if (route.embeddingMetadata === undefined) {
+      const inner = route.table.query();
+      inner.fullTextSearch({ query, columns });
+      return inner;
+    }
+
+    const metadata = route.embeddingMetadata;
+    if (cachedPreparation?.metadata !== metadata) {
+      cachedPreparation = {
+        metadata,
+        vector: Promise.resolve().then(() => getVector(metadata)),
+      };
+    }
+
+    const preparation = cachedPreparation;
+    let vector: Awaited<IntoVector>;
+    try {
+      vector = await preparation.vector;
+    } catch (error) {
+      if (cachedPreparation === preparation) {
+        cachedPreparation = undefined;
+      }
+      throw error;
+    }
+
+    return nearestToNative(route.table.query(), vector);
+  };
+
+  return new AutoQuery(createInner);
+}
+
 /**
 * A query that returns a subset of the rows in the table.
 *
@@ -836,37 +927,6 @@ export class Query extends StandardQueryBase<NativeQuery> {
    super(tbl.query());
  }

-  /** @hidden */
-  static autoSearch(
-    tbl: () => Promise<NativeTable>,
-    query: string,
-    vector: (tbl: NativeTable) => Promise<Awaited<IntoVector> | undefined>,
-    columns?: string[],
-  ): AutoQuery {
-    const nativeQuery = async () => {
-      const snapshot = await Promise.resolve(tbl());
-      const resolved = await vector(snapshot);
-      const inner = snapshot.query();
-      if (resolved === undefined) {
-        inner.fullTextSearch({
-          query,
-          columns: columns ?? null,
-        });
-        return inner;
-      }
-
-      const raw = Array.isArray(resolved)
-        ? null
-        : extractVectorBuffer(resolved);
-      if (raw) {
-        return inner.nearestToRaw(raw.data, raw.dtype);
-      }
-      return inner.nearestTo(Float32Array.from(resolved as number[]));
-    };
-
-    return new AutoQuery(nativeQuery);
-  }
-
  /**
   * Find the nearest vectors to the given query vector.
   *
@@ -905,45 +965,19 @@ export class Query extends StandardQueryBase<NativeQuery> {
   * a default `limit` of 10 will be used.  @see {@link Query#limit}
   */
  nearestTo(vector: IntoVector): VectorQuery {
-    const callNearestTo = (
-      inner: NativeQuery,
-      resolved: Float32Array | Float64Array | Uint8Array | number[],
-    ): NativeVectorQuery => {
-      const raw = Array.isArray(resolved)
-        ? null
-        : extractVectorBuffer(resolved);
-      if (raw) {
-        return inner.nearestToRaw(raw.data, raw.dtype);
-      }
-      return inner.nearestTo(Float32Array.from(resolved as number[]));
-    };
-
-    if (this.inner instanceof Promise) {
-      const nativeQuery = this.inner.then(async (inner) => {
-        const resolved = vector instanceof Promise ? await vector : vector;
-        return callNearestTo(inner, resolved);
-      });
+    const inner = this.inner;
+    if (inner instanceof Promise) {
+      const nativeQuery = inner.then(async (resolvedInner) =>
+        nearestToNative(resolvedInner, await vector),
+      );
      return new VectorQuery(nativeQuery);
    }
    if (vector instanceof Promise) {
-      const res = (async () => {
-        try {
-          const v = await vector;
-          // biome-ignore lint/suspicious/noExplicitAny: we need to get the `inner`, but js has no package scoping
-          const value: any = this.nearestTo(v);
-          const inner = value.inner as
-            | NativeVectorQuery
-            | Promise<NativeVectorQuery>;
-          return inner;
-        } catch (e) {
-          return Promise.reject(e);
-        }
-      })();
-      return new VectorQuery(res);
-    } else {
-      const vectorQuery = callNearestTo(this.inner, vector);
-      return new VectorQuery(vectorQuery);
+      return new VectorQuery(
+        vector.then((resolvedVector) => nearestToNative(inner, resolvedVector)),
+      );
    }
+    return new VectorQuery(nearestToNative(inner, vector));
  }

  nearestToText(query: string | FullTextQuery, columns?: string[]): Query {
@@ -48,6 +48,7 @@ import {
  Query,
  TakeQuery,
  VectorQuery,
+  createAutoQuery,
  instanceOfFullTextQuery,
 } from "./query";
 import { sanitizeType } from "./sanitize";
@@ -1177,33 +1178,29 @@ export class LocalTable extends Table {
      });
    }

-    if (queryType === "auto" && typeof query !== "string") {
-      return this.query().fullTextSearch(query, {
-        columns: ftsColumns,
-      });
-    }
+    if (queryType === "auto") {
+      if (instanceOfFullTextQuery(query)) {
+        return this.query().fullTextSearch(query, {
+          columns: ftsColumns,
+        });
+      }

-    if (queryType === "auto" && typeof query === "string") {
-      const vector = async (snapshot: _NativeTable) => {
-        const functions = await this.getEmbeddingFunctions(snapshot);
+      const columns =
+        typeof ftsColumns === "string" ? [ftsColumns] : (ftsColumns ?? null);
+      return createAutoQuery(this.inner, query, columns, async (metadata) => {
+        const functions = await getRegistry().parseFunctions(
+          new Map([["embedding_functions", metadata]]),
+        );
        // TODO: Support multiple embedding functions
        const embeddingFunc: EmbeddingFunctionConfig | undefined = functions
          .values()
          .next().value;
-        if (embeddingFunc === undefined) {
-          return undefined;
-        }
+        // The route only calls this callback when embedding metadata exists.
+        // parseFunctions either yields a provider or reports malformed metadata.
+        if (!embeddingFunc)
+          throw new Error("Invalid embedding function metadata");
        return await embeddingFunc.function.computeQueryEmbeddings(query);
-      };
-
-      const columns =
-        typeof ftsColumns === "string" ? [ftsColumns] : ftsColumns;
-      return Query.autoSearch(
-        () => this.inner.checkoutCurrent(),
-        query,
-        vector,
-        columns,
-      );
+      });
    }

    const queryPromise = this.getEmbeddingFunctions().then(
@@ -278,6 +278,13 @@ impl Table {
        Ok(Query::new(self.inner_ref()?.query()))
    }

+    /// Return a read-only table handle pinned to the current query revision.
+    #[napi(catch_unwind)]
+    pub async fn query_snapshot(&self) -> napi::Result<Self> {
+        let snapshot = self.inner_ref()?.query_snapshot().await.default_error()?;
+        Ok(Self::new(snapshot))
+    }
+
    #[napi(catch_unwind)]
    pub fn take_offsets(&self, offsets: Vec<i64>) -> napi::Result<TakeQuery> {
        Ok(TakeQuery::new(
@@ -47,6 +47,10 @@ impl TerminalResult {
        }
    }

+    pub(crate) fn value(&self) -> Option<&Value> {
+        self.value.as_ref()
+    }
+
    fn decode<T: DeserializeOwned>(self) -> Result<T> {
        let value = self.value.ok_or_else(|| match &self.request_id {
            Some(request_id) => Error::Http {
@@ -6,6 +6,7 @@
 use std::ops::Range;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;

 use arrow_array::{Array, LargeBinaryArray};
 use arrow_schema::DataType;
@@ -20,7 +21,7 @@ use crate::error::Result;
 use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient};
 use crate::table::BaseTable;

-use super::{FreshnessHeaders, RemoteTable};
+use super::{FreshnessHeaders, FreshnessState, RemoteTable, freshness_headers_snapshot};

 #[derive(Debug, Clone, Copy)]
 enum RangeRequestMode {
@@ -43,7 +44,10 @@ struct TableBlobRangeRequester<S: HttpSend> {
    path: String,
    version: Option<u64>,
    branch: Option<String>,
-    freshness: FreshnessHeaders,
+    freshness: Arc<std::sync::Mutex<FreshnessState>>,
+    parent_freshness: Arc<std::sync::Mutex<FreshnessState>>,
+    parent_freshness_request: FreshnessHeaders,
+    read_consistency_interval: Option<Duration>,
 }

 #[async_trait::async_trait]
@@ -53,8 +57,9 @@ impl<S: HttpSend> BlobRangeRequester for TableBlobRangeRequester<S> {
        range_header: &str,
        mode: RangeRequestMode,
    ) -> Result<(String, Response)> {
-        let mut request = self
-            .freshness
+        let freshness_request =
+            freshness_headers_snapshot(&self.freshness, self.read_consistency_interval);
+        let mut request = freshness_request
            .apply(self.client.get(&self.path))
            .header(header::RANGE, range_header);
        if let Some(version) = self.version {
@@ -71,6 +76,9 @@ impl<S: HttpSend> BlobRangeRequester for TableBlobRangeRequester<S> {
            return Ok((request_id, response));
        }
        let response = self.client.check_response(&request_id, response).await?;
+        freshness_request.observe_headers(&self.freshness, response.headers());
+        self.parent_freshness_request
+            .observe_headers(&self.parent_freshness, response.headers());
        Ok((request_id, response))
    }
 }
@@ -361,18 +369,21 @@ impl<S: HttpSend> RemoteTable<S> {
                message: "fetch_blobs is not supported on this LanceDB Cloud server".into(),
            });
        }
-        let version = self.current_version().await;
+        let read_snapshot = self.snapshot_read_state().await;
        let mut body = serde_json::json!({
-            "version": version,
+            "version": read_snapshot.version,
            "column": column,
            "row_ids": row_ids,
        });
        self.apply_branch_body(&mut body);

        let request = self
-            .post_read(&format!("/v1/table/{}/fetch_blobs/", self.identifier))
+            .client
+            .post(&format!("/v1/table/{}/fetch_blobs/", self.identifier))
            .json(&body);
-        let (request_id, response) = self.send(request, true).await?;
+        let (request_id, response) = self
+            .send_with_freshness(request, true, read_snapshot.freshness)
+            .await?;
        let mut stream = self.read_arrow_response(&request_id, response).await?;

        let mut blob_chunks: Vec<Arc<dyn Array>> = Vec::new();
@@ -448,8 +459,7 @@ impl<S: HttpSend> RemoteTable<S> {
            });
        }

-        let version = self.current_version().await;
-        let freshness = self.snapshot_freshness_headers();
+        let read_snapshot = self.snapshot_read_state().await;
        let encoded_column = urlencoding::encode(column);
        let requesters = row_ids
            .iter()
@@ -461,9 +471,12 @@ impl<S: HttpSend> RemoteTable<S> {
                let requester: Arc<dyn BlobRangeRequester> = Arc::new(TableBlobRangeRequester {
                    client: self.client.clone(),
                    path,
-                    version,
+                    version: read_snapshot.version,
                    branch: self.branch.clone(),
-                    freshness,
+                    freshness: Arc::new(std::sync::Mutex::new(read_snapshot.freshness_state)),
+                    parent_freshness: self.freshness.clone(),
+                    parent_freshness_request: read_snapshot.freshness,
+                    read_consistency_interval: self.client.read_consistency_interval,
                });
                requester
            })
@@ -685,6 +698,46 @@ mod tests {
        assert!(requests.lock().unwrap().contains(&"bytes=5-11".to_string()));
    }

+    #[tokio::test]
+    async fn remote_blob_file_keeps_the_open_timeline_after_parent_checkout() {
+        let range_requests = Arc::new(StdMutex::new(Vec::new()));
+        let captured = range_requests.clone();
+        let table = RemoteTable::new_mock(
+            "my_table".to_string(),
+            move |request| match request.url().path() {
+                "/v1/table/my_table/describe/" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"version":5,"schema":{"fields":[]}}"#.as_bytes().to_vec())
+                    .unwrap(),
+                "/v1/table/my_table/blob/image/10/bytes" => {
+                    captured.lock().unwrap().push((
+                        request.url().query().unwrap_or_default().to_string(),
+                        request.headers().clone(),
+                    ));
+                    range_response(&request, PAYLOAD)
+                }
+                path => panic!("unexpected path: {path}"),
+            },
+            Some(Version::new(0, 5, 0)),
+        );
+
+        table.checkout(5).await.unwrap();
+        let file = table
+            .fetch_blob_files_impl("image", &[10])
+            .await
+            .unwrap()
+            .pop()
+            .flatten()
+            .unwrap();
+        table.checkout_latest().await.unwrap();
+        file.read_range(5..12).await.unwrap();
+
+        let requests = range_requests.lock().unwrap();
+        let (query, headers) = requests.last().unwrap();
+        assert!(query.contains("version=5"));
+        assert!(!headers.contains_key("x-lancedb-min-timestamp"));
+    }
+
    #[tokio::test]
    async fn remote_blob_file_reuses_sequential_response_until_seek() {
        let requests = Arc::new(StdMutex::new(Vec::new()));
@@ -24,7 +24,10 @@ use lance::io::exec::utils::InstrumentedRecordBatchStreamAdapter;
 use crate::Error;
 use crate::remote::ARROW_STREAM_CONTENT_TYPE;
 use crate::remote::client::{HttpSend, RestfulLanceDbClient, Sender};
-use crate::remote::table::{MergeInsertRequest, REQUEST_TIMEOUT_HEADER, RemoteTable};
+use crate::remote::table::{
+    FreshnessHeaders, FreshnessState, MergeInsertRequest, REQUEST_TIMEOUT_HEADER, RemoteTable,
+    freshness_headers_snapshot,
+};
 use crate::table::datafusion::insert::COUNT_SCHEMA;
 use crate::table::write_progress::WriteProgressTracker;
 use crate::table::{AddResult, MergeResult};
@@ -54,6 +57,38 @@ pub enum WriteResult {
    Merge(MergeResult),
 }

+#[derive(Debug, Clone, Default)]
+struct WriteFreshness {
+    state: Option<Arc<Mutex<FreshnessState>>>,
+    read_consistency_interval: Option<Duration>,
+}
+
+impl WriteFreshness {
+    fn prepare(
+        &self,
+        request: reqwest::RequestBuilder,
+    ) -> (reqwest::RequestBuilder, Option<FreshnessHeaders>) {
+        match &self.state {
+            Some(state) => {
+                let freshness_request =
+                    freshness_headers_snapshot(state, self.read_consistency_interval);
+                (freshness_request.apply(request), Some(freshness_request))
+            }
+            None => (request, None),
+        }
+    }
+
+    fn observe(
+        &self,
+        freshness_request: Option<FreshnessHeaders>,
+        headers: &reqwest::header::HeaderMap,
+    ) {
+        if let (Some(state), Some(freshness_request)) = (&self.state, freshness_request) {
+            freshness_request.observe_headers(state, headers);
+        }
+    }
+}
+
 /// ExecutionPlan for streaming a write (add or merge_insert) to a remote
 /// LanceDB table.
 ///
@@ -71,6 +106,7 @@ pub struct RemoteWriteExec<S: HttpSend = Sender> {
    table_name: String,
    identifier: String,
    client: RestfulLanceDbClient<S>,
+    freshness: WriteFreshness,
    input: Arc<dyn ExecutionPlan>,
    op: WriteOp,
    properties: Arc<PlanProperties>,
@@ -170,6 +206,7 @@ impl<S: HttpSend + 'static> RemoteWriteExec<S> {
            table_name,
            identifier,
            client,
+            freshness: WriteFreshness::default(),
            input,
            op,
            properties: Arc::new(properties),
@@ -183,6 +220,18 @@ impl<S: HttpSend + 'static> RemoteWriteExec<S> {
        }
    }

+    pub(super) fn with_freshness(
+        mut self,
+        state: Arc<Mutex<FreshnessState>>,
+        read_consistency_interval: Option<Duration>,
+    ) -> Self {
+        self.freshness = WriteFreshness {
+            state: Some(state),
+            read_consistency_interval,
+        };
+        self
+    }
+
    /// Get the add result after execution, if this exec ran an insert.
    pub fn add_result(&self) -> Option<AddResult> {
        match self
@@ -285,6 +334,7 @@ impl<S: HttpSend + 'static> RemoteWriteExec<S> {
 /// each threading the same handful of arguments.
 struct PartRequestCtx<'a, S: HttpSend> {
    client: &'a RestfulLanceDbClient<S>,
+    freshness: &'a WriteFreshness,
    identifier: &'a str,
    table_name: &'a str,
    upload_id: &'a str,
@@ -352,7 +402,11 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
    }

    /// Build the `/insert` request for a single multipart part.
-    fn build_part_request(&self, part_id: &str, body: reqwest::Body) -> reqwest::RequestBuilder {
+    fn build_part_request(
+        &self,
+        part_id: &str,
+        body: reqwest::Body,
+    ) -> (reqwest::RequestBuilder, Option<FreshnessHeaders>) {
        let mut request = self
            .client
            .post(&format!("/v1/table/{}/insert/", self.identifier))
@@ -368,12 +422,16 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
        if let Some(b) = self.branch {
            request = request.query(&[("branch", b)]);
        }
-        request.body(body)
+        self.freshness.prepare(request.body(body))
    }

    /// Send a single part's request and drain the response, mapping HTTP and
    /// table-not-found errors into `DataFusionError`.
-    async fn send_part_request(&self, request: reqwest::RequestBuilder) -> DataFusionResult<()> {
+    async fn send_part_request(
+        &self,
+        request: reqwest::RequestBuilder,
+        freshness_request: Option<FreshnessHeaders>,
+    ) -> DataFusionResult<()> {
        let (request_id, response) = self
            .client
            .send(request)
@@ -388,6 +446,8 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
            .check_response(&request_id, response)
            .await
            .map_err(|e| DataFusionError::External(Box::new(e)))?;
+        self.freshness
+            .observe(freshness_request, response.headers());
        response.bytes().await.map_err(|e| {
            DataFusionError::External(Box::new(Error::Http {
                source: Box::new(e),
@@ -419,7 +479,7 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
        let body = reqwest::Body::wrap_stream(chunk_rx);

        let part_id = uuid::Uuid::new_v4().to_string();
-        let request = self.build_part_request(&part_id, body);
+        let (request, freshness_request) = self.build_part_request(&part_id, body);

        // Measured from just before the request is sent, matching the window the
        // client read timeout applies to the upload.
@@ -495,7 +555,7 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
            Ok::<bool, DataFusionError>(input_ended)
        };

-        let send = self.send_part_request(request);
+        let send = self.send_part_request(request, freshness_request);

        // `join!` rather than `tokio::spawn`: the producer borrows `input` (and
        // `schema`), so it cannot satisfy the `'static` bound a spawned task
@@ -569,7 +629,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
        // Building a fresh exec (with a new, empty `result`) is what makes the
        // outer rescannable retry loop work: `reset_state()` clears the captured
        // result so a re-execution starts clean.
-        Ok(Arc::new(Self::new_inner(
+        let mut exec = Self::new_inner(
            self.table_name.clone(),
            self.identifier.clone(),
            self.client.clone(),
@@ -580,7 +640,9 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
            self.branch.clone(),
            self.max_bytes_per_request,
            self.max_request_duration,
-        )))
+        );
+        exec.freshness = self.freshness.clone();
+        Ok(Arc::new(exec))
    }

    fn execute(
@@ -613,6 +675,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
                &self.metrics,
            ));
        let client = self.client.clone();
+        let freshness = self.freshness.clone();
        let identifier = self.identifier.clone();
        let op = self.op.clone();
        let result_slot = self.result.clone();
@@ -634,6 +697,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
                let overwrite = matches!(op, WriteOp::Insert { overwrite: true });
                let ctx = PartRequestCtx {
                    client: &client,
+                    freshness: &freshness,
                    identifier: &identifier,
                    table_name: &table_name,
                    upload_id,
@@ -688,7 +752,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {

            let (error_tx, mut error_rx) = tokio::sync::oneshot::channel();
            let body = Self::stream_as_http_body(input_stream, error_tx, tracker)?;
-            let request = request.body(body);
+            let (request, freshness_request) = freshness.prepare(request.body(body));

            let result: DataFusionResult<(String, _)> = async {
                let (request_id, response) = client
@@ -708,6 +772,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
                    .check_response(&request_id, response)
                    .await
                    .map_err(|e| DataFusionError::External(Box::new(e)))?;
+                freshness.observe(freshness_request, response.headers());

                Ok((request_id, response))
            }
@@ -560,6 +560,13 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
    fn id(&self) -> &str;
    /// Get the arrow [Schema] of the table.
    async fn schema(&self) -> Result<SchemaRef>;
+    /// Create a read-only handle pinned to the table's current active revision.
+    ///
+    /// The returned handle is independent from later refreshes or checkouts on
+    /// this handle.  This is used by bindings that must prepare client-side
+    /// query state from the same revision that the query will execute against.
+    #[doc(hidden)]
+    async fn query_snapshot(&self) -> Result<Arc<dyn BaseTable>>;
    /// Count the number of rows in this table.
    async fn count_rows(&self, filter: Option<Filter>) -> Result<usize>;
    /// Create a physical plan for the query.
@@ -1139,6 +1146,16 @@ impl Table {
        self.inner.schema().await
    }

+    /// Create a read-only handle pinned to the current active revision.
+    #[doc(hidden)]
+    pub async fn query_snapshot(&self) -> Result<Self> {
+        Ok(Self {
+            inner: self.inner.query_snapshot().await?,
+            database: self.database.clone(),
+            embedding_registry: self.embedding_registry.clone(),
+        })
+    }
+
    /// Count the number of rows in this dataset.
    ///
    /// # Arguments
@@ -3059,6 +3076,17 @@ impl BaseTable for NativeTable {
        &self.id
    }

+    async fn query_snapshot(&self) -> Result<Arc<dyn BaseTable>> {
+        let snapshot = self.dataset.new_query_snapshot().await?;
+        let mut table = self.with_dataset(snapshot);
+        // QueryTable requests do not carry a revision. A pinned snapshot must
+        // execute locally until the namespace API can accept that revision.
+        table
+            .pushdown_operations
+            .remove(&NamespaceClientPushdownOperation::QueryTable);
+        Ok(Arc::new(table))
+    }
+
    async fn version(&self) -> Result<u64> {
        Ok(self.dataset.get().await?.version().version)
    }
@@ -22,7 +22,7 @@
 use std::collections::{BTreeSet, HashMap};
 use std::sync::Arc;

-use arrow_schema::{DataType, Field as ArrowField, Fields, Schema as ArrowSchema, SchemaRef};
+use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema, SchemaRef};
 use datafusion_common::tree_node::TreeNode;
 use datafusion_physical_plan::PhysicalExpr;
 use lance::dataset::NewColumnTransform;
@@ -1273,11 +1273,6 @@ pub(crate) fn bind(schema: SchemaRef, column: &str, expression: &str) -> Result<
 /// refresh time: that the expression parses, that every column it reads
 /// exists, and that the target name is free. A declaration that survives this
 /// is one a refresh can always act on.
-///
-/// Each accepted column joins the schema the next one resolves against, so a
-/// batch may declare `a` and then `b = a + 1` in one commit. Refresh order
-/// then matters, and refresh enforces it: `b` is refused while `a` still has
-/// unfilled rows.
 pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Vec<ArrowField>> {
    if columns.is_empty() {
        return Err(Error::InvalidInput {
@@ -1285,11 +1280,11 @@ pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Ve
        });
    }

-    let mut schema = schema;
    let mut fields = Vec::with_capacity(columns.len());
+    let mut declared: Vec<&str> = Vec::with_capacity(columns.len());

    for (name, expression) in columns {
-        if schema.field_with_name(name).is_ok() {
+        if schema.field_with_name(name).is_ok() || declared.contains(&name.as_str()) {
            return Err(Error::ColumnAlreadyExists { name: name.clone() });
        }

@@ -1297,50 +1292,16 @@ pub(crate) fn plan(schema: SchemaRef, columns: &[(String, String)]) -> Result<Ve

        // Declared columns start entirely null, so nullability is a property
        // of the declaration rather than of what the expression yields.
-        let field = ArrowField::new(name, bound.data_type, true)
-            .with_metadata(computed_column_metadata(expression, &bound.inputs));
-        schema = Arc::new(ArrowSchema::new_with_metadata(
-            schema
-                .fields()
-                .iter()
-                .cloned()
-                .chain(std::iter::once(Arc::new(field.clone())))
-                .collect::<Fields>(),
-            schema.metadata().clone(),
-        ));
-        fields.push(field);
+        fields.push(
+            ArrowField::new(name, bound.data_type, true)
+                .with_metadata(computed_column_metadata(expression, &bound.inputs)),
+        );
+        declared.push(name);
    }

    Ok(fields)
 }

-/// Run the schema-level checks of
-/// [`AddColumnsBuilder::computed`](super::AddColumnsBuilder::computed) against
-/// `schema` without committing: the Function-binding guard and the planning of
-/// every declaration. For callers that stage declarations behind other work
-/// and need those rejections before any of it lands.
-///
-/// Only the schema is consulted. Declaring also refuses a table with an LSM
-/// write spec or retained SSTables; that is table state, checked at commit.
-///
-/// ```
-/// # use std::sync::Arc;
-/// # use arrow_schema::{DataType, Field, Schema};
-/// use lancedb::table::computed_columns::validate_declarations;
-///
-/// let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
-/// let declarations = vec![
-///     ("a".to_string(), "x + 1".to_string()),
-///     ("b".to_string(), "a * 2".to_string()),
-/// ];
-/// assert!(validate_declarations(schema.clone(), &declarations).is_ok());
-/// assert!(validate_declarations(schema, &[("c".into(), "random()".into())]).is_err());
-/// ```
-pub fn validate_declarations(schema: SchemaRef, columns: &[(String, String)]) -> Result<()> {
-    ensure_no_function_bindings_for_mutation(schema.as_ref(), "schema evolution")?;
-    plan(schema, columns).map(drop)
-}
-
 /// Build the transform that declares `columns` against `schema`.
 ///
 /// An all-null column is how a binding with no values yet is carried into a
@@ -1379,22 +1340,6 @@ pub(super) async fn add_foreign_kind(table: &crate::Table, name: &str, kind: &st

 #[cfg(test)]
 mod tests {
-    /// The gate's reproducer: the validator applies the same schema-level
-    /// guard declaring does, so a staging caller is refused before it commits
-    /// anything else.
-    #[test]
-    fn test_validate_declarations_matches_schema_admission_barriers() {
-        let schema = Arc::new(ArrowSchema::new_with_metadata(
-            vec![ArrowField::new("x", DataType::Int32, true)],
-            HashMap::from([(
-                FUNCTION_BINDINGS_META_KEY.to_string(),
-                "not valid binding metadata".to_string(),
-            )]),
-        ));
-        let declarations = vec![("a".to_string(), "x + 1".to_string())];
-        assert!(super::validate_declarations(schema, &declarations).is_err());
-    }
-
    #[test]
    fn output_arrow_type_grammar_matches_the_shared_golden() {
        let golden: serde_json::Value = serde_json::from_str(include_str!(
@@ -1637,40 +1582,6 @@ mod tests {
        assert!(declared(&table).await.is_empty());
    }

-    /// A batch may build on itself: one commit, and the later entry's inputs
-    /// name the earlier one.
-    #[tokio::test]
-    async fn test_a_declaration_may_read_one_declared_before_it() {
-        let table = table_with_ints("chain").await;
-        let before = table.version().await.unwrap();
-        add_computed(
-            &table,
-            &[("a".into(), "x + 1".into()), ("b".into(), "a * 2".into())],
-        )
-        .await
-        .unwrap();
-        assert_eq!(table.version().await.unwrap(), before + 1);
-        let declared = declared(&table).await;
-        assert_eq!(declared[1].name, "b");
-        assert_eq!(declared[1].inputs, vec!["a".to_string()]);
-
-        // Order is the dependency order; reading ahead is still unknown.
-        let err = add_computed(
-            &table,
-            &[("c".into(), "d + 1".into()), ("d".into(), "x + 1".into())],
-        )
-        .await
-        .unwrap_err();
-        assert!(matches!(err, Error::InvalidExpression { column, .. } if column == "c"));
-        assert!(
-            validate_declarations(
-                table.schema().await.unwrap(),
-                &[("e".into(), "random()".into())]
-            )
-            .is_err()
-        );
-    }
-
    /// A column added by an ordinary transform is materialized, not bound, so
    /// it carries no declaration to report.
    #[tokio::test]
@@ -32,6 +32,10 @@ struct DatasetState {
    /// `Some(version)` = pinned to a specific version (time travel),
    /// `None` = tracking latest.
    pinned_version: Option<u64>,
+    /// Whether the pin is an internal query snapshot rather than user-visible
+    /// time travel. Query snapshots remain read-only but preserve MemWAL read
+    /// semantics.
+    query_snapshot: bool,
 }

 #[derive(Debug, Clone)]
@@ -70,6 +74,7 @@ impl DatasetConsistencyWrapper {
            state: Arc::new(Mutex::new(DatasetState {
                dataset,
                pinned_version: None,
+                query_snapshot: false,
            })),
            consistency,
            shard_writer: Arc::new(ShardWriterCache::default()),
@@ -93,6 +98,36 @@ impl DatasetConsistencyWrapper {
        wrapper
    }

+    /// Create an independent read-only wrapper pinned to the current dataset
+    /// while retaining this wrapper's live MemWAL read context.
+    pub async fn new_query_snapshot(&self) -> Result<Self> {
+        // Apply the configured consistency policy before taking the snapshot.
+        // The returned dataset is intentionally discarded: a checkout may race
+        // after this await, so the dataset and its pin provenance must instead
+        // be cloned together from one authoritative state sample below.
+        self.get().await?;
+
+        let (dataset, query_snapshot) = {
+            let state = self.state.lock()?;
+            // Preserve user time travel so the MemWAL safety guard still sees
+            // it. Latest and already-internal snapshots remain internal pins.
+            (
+                state.dataset.clone(),
+                state.query_snapshot || state.pinned_version.is_none(),
+            )
+        };
+        let version = dataset.version().version;
+        Ok(Self {
+            state: Arc::new(Mutex::new(DatasetState {
+                dataset,
+                pinned_version: Some(version),
+                query_snapshot,
+            })),
+            consistency: ConsistencyMode::Lazy,
+            shard_writer: self.shard_writer.clone(),
+        })
+    }
+
    /// The MemWAL `ShardWriter` cache co-located with this dataset.
    pub(crate) fn shard_writer(&self) -> &Arc<ShardWriterCache> {
        &self.shard_writer
@@ -169,6 +204,7 @@ impl DatasetConsistencyWrapper {
        let mut state = self.state.lock()?;
        state.dataset = Arc::new(new_dataset);
        state.pinned_version = None;
+        state.query_snapshot = false;
        drop(state);
        if let ConsistencyMode::Eventual(bg_cache) = &self.consistency {
            bg_cache.invalidate();
@@ -202,10 +238,10 @@ impl DatasetConsistencyWrapper {

    /// Returns the version, if in time travel mode, or None otherwise.
    pub fn time_travel_version(&self) -> Option<u64> {
-        self.state
-            .lock()
-            .unwrap_or_else(|e| e.into_inner())
-            .pinned_version
+        let state = self.state.lock().unwrap_or_else(|e| e.into_inner());
+        (!state.query_snapshot)
+            .then_some(state.pinned_version)
+            .flatten()
    }

    /// Convert into a wrapper in latest version mode.
@@ -225,6 +261,7 @@ impl DatasetConsistencyWrapper {
        if state.pinned_version.is_some() {
            state.dataset = Arc::new(new_dataset);
            state.pinned_version = None;
+            state.query_snapshot = false;
        }
        drop(state);
        if let ConsistencyMode::Eventual(bg_cache) = &self.consistency {
@@ -260,6 +297,7 @@ impl DatasetConsistencyWrapper {
        let mut state = self.state.lock()?;
        state.dataset = Arc::new(new_dataset);
        state.pinned_version = Some(version_value);
+        state.query_snapshot = false;
        Ok(())
    }

@@ -461,6 +499,29 @@ mod tests {
        assert_eq!(wrapper.time_travel_version(), Some(1));
    }

+    #[tokio::test]
+    async fn test_query_snapshot_samples_dataset_and_pin_together() {
+        let dir = tempfile::tempdir().unwrap();
+        let uri = dir.path().to_str().unwrap();
+        let ds = create_test_dataset(uri).await;
+
+        let wrapper = DatasetConsistencyWrapper::new_latest(ds, None);
+        wrapper.as_time_travel(1u64).await.unwrap();
+        let stale_time_travel_dataset = wrapper.get().await.unwrap();
+
+        append_to_dataset(uri).await;
+        wrapper.as_latest().await.unwrap();
+
+        let snapshot = wrapper.new_query_snapshot().await.unwrap();
+        let snapshot_dataset = snapshot.get().await.unwrap();
+        assert_eq!(snapshot_dataset.version().version, 2);
+        assert_ne!(
+            snapshot_dataset.version().version,
+            stale_time_travel_dataset.version().version
+        );
+        assert_eq!(snapshot.time_travel_version(), None);
+    }
+
    #[tokio::test]
    async fn test_as_latest_from_time_travel() {
        let dir = tempfile::tempdir().unwrap();
@@ -1056,6 +1056,44 @@ mod lsm_tests {
        );
    }

+    #[tokio::test]
+    async fn query_snapshot_preserves_lsm_read_semantics() {
+        let dir = tempdir().unwrap();
+        let table = id_value_table(&dir).await;
+        table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded())
+            .await
+            .unwrap();
+        lsm_upsert(&table, vec![4, 5]).await;
+
+        let snapshot = table.query_snapshot().await.unwrap();
+        let rows = collect_id_value(snapshot.query().execute().await.unwrap()).await;
+        assert_eq!(
+            rows.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
+            vec![1, 2, 3, 4, 5]
+        );
+    }
+
+    #[tokio::test]
+    async fn query_snapshot_preserves_time_travel_lsm_guard() {
+        let dir = tempdir().unwrap();
+        let table = id_value_table(&dir).await;
+        table
+            .set_lsm_write_spec(LsmWriteSpec::unsharded())
+            .await
+            .unwrap();
+        lsm_upsert(&table, vec![4]).await;
+
+        let version = table.version().await.unwrap();
+        table.checkout(version).await.unwrap();
+        let direct_error = table.query().execute().await.err().unwrap();
+        assert!(matches!(direct_error, Error::NotSupported { .. }));
+
+        let snapshot = table.query_snapshot().await.unwrap();
+        let snapshot_error = snapshot.query().execute().await.err().unwrap();
+        assert!(matches!(snapshot_error, Error::NotSupported { .. }));
+    }
+
    #[tokio::test]
    async fn lsm_read_dedup_newest_wins() {
        let dir = tempdir().unwrap();
@@ -1056,6 +1056,37 @@ mod tests {
        assert_eq!(namespace_client.query_table_calls.load(Ordering::SeqCst), 0);
    }

+    #[tokio::test]
+    async fn test_query_snapshot_disables_namespace_pushdown() {
+        use crate::connect;
+        use crate::table::BaseTable;
+        use arrow_array::{Int32Array, RecordBatch};
+        use arrow_schema::{DataType, Field, Schema};
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch =
+            RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1, 2, 3]))]).unwrap();
+        let table = conn
+            .create_table("test_snapshot_namespace_fallback", vec![batch])
+            .execute()
+            .await
+            .unwrap();
+        let mut native_table = table.as_native().unwrap().clone();
+        native_table.namespace_client = Some(Arc::new(CountingNamespaceClient::default()));
+        native_table
+            .pushdown_operations
+            .insert(NamespaceClientPushdownOperation::QueryTable);
+
+        let snapshot = BaseTable::query_snapshot(&native_table).await.unwrap();
+        let snapshot = snapshot.as_any().downcast_ref::<NativeTable>().unwrap();
+        assert!(
+            !can_execute_namespace_query(snapshot, &AnyQuery::Query(QueryRequest::default()),)
+                .await
+                .unwrap()
+        );
+    }
+
    #[tokio::test]
    async fn test_create_plan_multivector_structure() {
        use arrow_array::{Float32Array, RecordBatch};
@@ -7,16 +7,6 @@
 //! therefore idempotent and does not observe input mutation -- once a row is
 //! filled, changing what the expression reads leaves the stored result alone.
 //!
-//! A column's computed inputs are filled first -- the dependency graph is
-//! walked once, each reachable column filled once in dependency order, each
-//! fill its own commit. Every fill in the pass, the requested column's
-//! included, covers only the fragments of the snapshot the pass started
-//! from: a commit may rebase over a concurrent append, and the fragment that
-//! admits carries placeholder nulls no earlier fill covered, so it waits for
-//! a later refresh rather than being read as values. Two concurrent fills of
-//! one input collide on its field in lance's conflict check, so a dependent
-//! fill can only commit over inputs that were durable when it read them.
-//!
 //! Two passes per fragment. The first scans only the unfilled live rows and
 //! evaluates the expression over them, which yields the exact fill count and
 //! decides whether the fragment is staged at all -- a fragment where nothing
@@ -51,8 +41,7 @@ use crate::{Error, Result};
 /// The result of refreshing a computed column.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
 pub struct RefreshColumnResult {
-    /// Rows that had a value computed, in the requested column only; inputs
-    /// filled on its behalf are not counted.
+    /// Rows that had a value computed.
    #[serde(default)]
    pub rows_filled: u64,
    /// The commit version associated with the operation.
@@ -63,7 +52,6 @@ pub struct RefreshColumnResult {
 struct RefreshExecution {
    result: RefreshColumnResult,
    source_version: u64,
-    published_version: Option<u64>,
 }

 /// Internal implementation of the refresh logic.
@@ -86,12 +74,7 @@ async fn execute_refresh_column_with_source(

    let expression = declared_expression(&dataset, column)?;
    let schema = Arc::new(ArrowSchema::from(dataset.schema()));
-    let bound = Arc::new(super::computed_columns::bind(
-        schema.clone(),
-        column,
-        &expression,
-    )?);
-    ensure_inputs_filled(&dataset, &schema, column, &bound).await?;
+    let bound = Arc::new(super::computed_columns::bind(schema, column, &expression)?);
    let field = dataset
        .schema()
        .field(column)
@@ -117,25 +100,25 @@ async fn execute_refresh_column_with_source(
        replacements.push(fragment.write_columns(values, &column_schema).await?);
    }

-    let source_version = dataset.version().version;
    if replacements.is_empty() {
+        let source_version = dataset.version().version;
        return Ok(RefreshExecution {
            result: RefreshColumnResult {
                rows_filled: 0,
                version: source_version,
            },
            source_version,
-            published_version: None,
        });
    }

+    let read_version = dataset.version().version;
    // The dataset's own session, so registrations and caches survive the
    // commit being installed on the handle.
    let session = dataset.session();
    let new_dataset = Dataset::commit(
        WriteDestination::Dataset(dataset.clone()),
        Operation::DataReplacement { replacements },
-        Some(source_version),
+        Some(read_version),
        None,
        None,
        session,
@@ -150,52 +133,10 @@ async fn execute_refresh_column_with_source(
            rows_filled,
            version,
        },
-        source_version,
-        published_version: Some(version),
+        source_version: read_version,
    })
 }

-/// Refuse while a computed input still has rows a refresh of it would fill:
-/// read now, its placeholder null would be evaluated as a value and kept.
-async fn ensure_inputs_filled(
-    dataset: &Dataset,
-    schema: &Arc<ArrowSchema>,
-    column: &str,
-    bound: &BoundExpression,
-) -> Result<()> {
-    for input in &bound.roots {
-        let Some(declaration) = schema
-            .field_with_name(input)
-            .ok()
-            .and_then(computed_column_from_field)
-        else {
-            continue;
-        };
-        let ComputedColumnKind::Sql { expression } = &declaration.kind else {
-            return Err(Error::NotSupported {
-                message: format!(
-                    "computed column '{column}' reads '{input}', whose fill state this \
-                     refresh cannot check; refresh '{input}' first"
-                ),
-            });
-        };
-        let input_bound = super::computed_columns::bind(schema.clone(), input, expression)?;
-        let mut unfilled = 0u64;
-        for fragment in dataset.get_fragments() {
-            unfilled += count_fragment_gains(dataset, &fragment, &input_bound, input).await?;
-        }
-        if unfilled > 0 {
-            return Err(Error::InvalidInput {
-                message: format!(
-                    "computed column '{column}' reads '{input}', which has {unfilled} unfilled \
-                     rows; refresh '{input}' first"
-                ),
-            });
-        }
-    }
-    Ok(())
-}
-
 /// Run the refresh as a [`Job`] in this process.
 pub(crate) async fn execute_refresh_column_async(
    table: &NativeTable,
@@ -219,7 +160,8 @@ pub(crate) async fn execute_refresh_column_async(
            rows_failed: 0,
            rows_remaining: 0,
            source_version: execution.source_version,
-            published_version: execution.published_version,
+            published_version: (execution.result.rows_filled > 0)
+                .then_some(execution.result.version),
        })
    })))
 }
@@ -442,8 +384,7 @@ mod tests {
            .version)
    }

-    async fn read(table: &Table, column: &str) -> Vec<Option<i64>> {
-        use arrow_array::{Array, Int64Array};
+    async fn read(table: &Table, column: &str) -> Vec<Option<i32>> {
        let batches = table
            .query()
            .select(Select::columns(&[column]))
@@ -453,19 +394,15 @@ mod tests {
            .try_collect::<Vec<_>>()
            .await
            .unwrap();
-        let mut values: Vec<Option<i64>> = batches
+        let mut values: Vec<Option<i32>> = batches
            .iter()
            .flat_map(|batch| {
-                let array = &batch[column];
-                match array.as_any().downcast_ref::<Int32Array>() {
-                    Some(ints) => ints.iter().map(|v| v.map(i64::from)).collect::<Vec<_>>(),
-                    None => array
-                        .as_any()
-                        .downcast_ref::<Int64Array>()
-                        .unwrap()
-                        .iter()
-                        .collect::<Vec<_>>(),
-                }
+                batch[column]
+                    .as_any()
+                    .downcast_ref::<Int32Array>()
+                    .unwrap()
+                    .iter()
+                    .collect::<Vec<_>>()
            })
            .collect();
        values.sort();
@@ -477,98 +414,6 @@ mod tests {
        table.add(batch).execute().await.unwrap();
    }

-    /// The gate's reproducer: `b = coalesce(a, 0)` refreshed before `a`
-    /// must not bake zeros from `a`'s placeholder null. It is refused, and
-    /// names the input, until `a` is filled -- after every append too.
-    #[tokio::test]
-    async fn test_dependent_refresh_refuses_an_unfilled_input() {
-        let table = table_with("dependent_refresh_order", vec![1, 2, 3]).await;
-        table
-            .add_columns()
-            .computed("a", "x + 1")
-            .computed("b", "coalesce(a, 0)")
-            .execute()
-            .await
-            .unwrap();
-
-        let err = table.refresh_column("b").await.unwrap_err();
-        assert!(
-            matches!(&err, Error::InvalidInput { message } if message.contains("refresh 'a' first")),
-            "{err}"
-        );
-        assert_eq!(read(&table, "b").await, vec![None, None, None]);
-
-        assert_eq!(table.refresh_column("a").await.unwrap().rows_filled, 3);
-        assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 3);
-        assert_eq!(read(&table, "b").await, vec![Some(2), Some(3), Some(4)]);
-
-        append(&table, vec![10]).await;
-        assert!(table.refresh_column("b").await.is_err());
-        table.refresh_column("a").await.unwrap();
-        assert_eq!(table.refresh_column("b").await.unwrap().rows_filled, 1);
-        assert_eq!(
-            table.count_rows(Some("b = 0".to_string())).await.unwrap(),
-            0
-        );
-    }
-
-    /// Names that need quoting, and a nested input, survive the trip through
-    /// declaration metadata and the dependency check: the recorded inputs
-    /// are matched by name, never re-parsed as SQL.
-    #[tokio::test]
-    async fn test_dependent_refresh_handles_awkward_column_names() {
-        use arrow_array::{Int32Array, StructArray};
-        use arrow_schema::{DataType, Field, Fields};
-
-        let conn = connect("memory://").execute().await.unwrap();
-        let age_fields = Fields::from(vec![Field::new("age", DataType::Int32, true)]);
-        let meta = StructArray::new(
-            age_fields.clone(),
-            vec![Arc::new(Int32Array::from(vec![10, 20])) as _],
-            None,
-        );
-        let schema = Arc::new(arrow_schema::Schema::new(vec![
-            Field::new("camelCase", DataType::Int32, true),
-            Field::new("with-hyphen", DataType::Int32, true),
-            Field::new("meta", DataType::Struct(age_fields), true),
-        ]));
-        let batch = arrow_array::RecordBatch::try_new(
-            schema,
-            vec![
-                Arc::new(Int32Array::from(vec![1, 2])) as _,
-                Arc::new(Int32Array::from(vec![100, 200])) as _,
-                Arc::new(meta) as _,
-            ],
-        )
-        .unwrap();
-        let table = conn
-            .create_table("awkward_names", batch)
-            .execute()
-            .await
-            .unwrap();
-
-        table
-            .add_columns()
-            .computed("y", "`camelCase` * 2")
-            .computed("z", "coalesce(y, 0) + `with-hyphen` + meta.age")
-            .execute()
-            .await
-            .unwrap();
-        let z = crate::table::computed_columns::computed_columns(
-            table.schema().await.unwrap().as_ref(),
-        )
-        .into_iter()
-        .find(|c| c.name == "z")
-        .unwrap();
-        assert_eq!(z.inputs, vec!["meta.age", "with-hyphen", "y"]);
-
-        let err = table.refresh_column("z").await.unwrap_err();
-        assert!(err.to_string().contains("refresh 'y' first"), "{err}");
-        assert_eq!(table.refresh_column("y").await.unwrap().rows_filled, 2);
-        assert_eq!(table.refresh_column("z").await.unwrap().rows_filled, 2);
-        assert_eq!(read(&table, "z").await, vec![Some(112), Some(224)]);
-    }
-
    #[tokio::test]
    async fn test_refresh_fills_a_declared_column() {
        let table = table_with("refresh_fills", vec![1, 2, 3]).await;
@@ -806,8 +651,7 @@ mod tests {

        let read_back = read(&table, "doubled").await;
        assert_eq!(read_back.len(), 20_000);
-        let mut expected: Vec<Option<i64>> =
-            values.iter().map(|v| Some(i64::from(v * 2))).collect();
+        let mut expected: Vec<Option<i32>> = values.iter().map(|v| Some(v * 2)).collect();
        expected.sort();
        assert_eq!(read_back, expected);
    }
Author	SHA1	Message	Date
Jack Ye	b81e5c2971	chore: use streamed-copy lance branch	2026-08-25 22:46:47 -07:00
Jack Ye	391cac9034	fix(remote): centralize timeline consistency (#4053 ) Centralizes remote table freshness fencing and response-version tracking in the default transport path. Covers schema and blob bypass paths, keeps explicit time-travel and cross-timeline operations unfenced, and advances freshness after refresh and index job completion.	2026-08-26 12:54:36 +08:00
LanceDB Robot	21530432a0	chore: update lance dependency to v12.0.0-beta.2 (#4056 ) Updates the Rust workspace Lance crates and Java lance-core dependency to [v12.0.0-beta.2](https://github.com/lance-format/lance/releases/tag/v12.0.0-beta.2). No compatibility fixes were required; full-workspace Clippy passes with all features and warnings denied.	2026-08-25 21:56:26 -05:00
lancedb-gatefixer[bot]	9b825c5f29	fix(node): route auto search using table embeddings (#3832 ) ## Summary - Resolve automatic string-search routing from the active table schema whenever the query executes. - Defer embedding-provider construction while leaving explicit vector and FTS routes unchanged. - Cover unrelated global registrations and metadata transitions across repeated executions of one query builder. ## Root cause LocalTable.search used the number of globally registered embedding providers to choose between vector and full-text search. A provider registered for any other table therefore sent a plain FTS table down the vector path. A wrapper-lifetime metadata snapshot avoided that contamination but became stale after time travel or read-consistency refreshes. The query now records fluent builder operations and creates the appropriate native vector or FTS query from the active schema on each execution. ## Validation - pnpm build - pnpm tsc - pnpm lint - pnpm run docs - pnpm test --runInBand (681 passed, 5 skipped) Fixes #1557 <!-- lance-gatekeeper-fix:v1 agent=b6183df8296db4aabdc5d19a2256b029 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-26 10:08:48 +08:00