Merge remote-tracking branch 'refs/remotes/origin/main' into gatekeeper/fix-2820-1

# Conflicts: # rust/lancedb/src/remote/table.rs
fix(remote): centralize timeline consistency (#4053 )
2026-08-26 07:58:31 +00:00 · 2026-08-26 05:43:48 +00: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 · 2026-08-25 16:49:17 -07:00
38 changed files with 4371 additions and 529 deletions
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4815,8 +4815,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4888,8 +4888,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 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=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 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=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4934,8 +4934,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4945,8 +4945,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4983,8 +4983,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5013,8 +5013,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5031,8 +5031,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5041,8 +5041,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5075,8 +5075,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5107,8 +5107,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5172,8 +5172,8 @@ dependencies = [

 [[package]]
 name = "lance-index-core"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5195,8 +5195,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5236,8 +5236,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5251,8 +5251,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5264,8 +5264,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5318,8 +5318,8 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5333,8 +5333,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5374,8 +5374,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5388,8 +5388,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "11.0.0-beta.22"
-source = "git+https://github.com/lance-format/lance.git?tag=v11.0.0-beta.22#ea3cb4d799c468232735e9bcb43959487aca5c20"
+version = "12.0.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.2#dafa4642658d996b3e31dde91e02f72db7860d7e"
 dependencies = [
 "frostem",
 "icu_segmenter",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=11.0.0-beta.22", default-features = false, "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=11.0.0-beta.22", default-features = false, "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=11.0.0-beta.22", default-features = false, "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=11.0.0-beta.22", "tag" = "v11.0.0-beta.22", "git" = "https://github.com/lance-format/lance.git" }
+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" }
 lancedb = { path = "rust/lancedb", default-features = false }
 ahash = "0.8"
 # Note that this one does not include pyarrow
@@ -0,0 +1,518 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / AutoQuery
+
+# Class: AutoQuery
+
+A builder for automatic string searches.
+
+Automatic search determines whether to use full-text or vector search from
+the table revision selected for each execution. This builder exposes the
+common operations supported by both query families.
+
+## Extends
+
+- `StandardQueryBase`&lt;`NativeQuery` \| `NativeVectorQuery`&gt;
+
+## Properties
+
+### inner
+
+```ts
+protected inner: Query | VectorQuery | Promise<Query | VectorQuery>;
+```
+
+#### Inherited from
+
+`StandardQueryBase.inner`
+
+## Methods
+
+### analyzePlan()
+
+```ts
+analyzePlan(distributedMetrics?): Promise<string>
+```
+
+Executes the query and returns the physical query plan annotated with runtime metrics.
+
+This is useful for debugging and performance analysis, as it shows how the query was executed
+and includes metrics such as elapsed time, rows processed, and I/O statistics.
+
+#### Parameters
+
+* **distributedMetrics?**: [`AnalyzePlanDistributedMetrics`](../type-aliases/AnalyzePlanDistributedMetrics.md)
+    How distributed worker metrics are displayed for remote query plans.
+    Defaults to `"aggregate"`.
+
+#### Returns
+
+`Promise`&lt;`string`&gt;
+
+A query execution plan with runtime metrics for each step.
+
+#### Example
+
+```ts
+import * as lancedb from "@lancedb/lancedb"
+
+const db = await lancedb.connect("./.lancedb");
+const table = await db.createTable("my_table", [
+  { vector: [1.1, 0.9], id: "1" },
+]);
+
+const plan = await table.query().nearestTo([0.5, 0.2]).analyzePlan();
+
+Example output (with runtime metrics inlined):
+AnalyzeExec verbose=true, metrics=[]
+ ProjectionExec: expr=[id@3 as id, vector@0 as vector, _distance@2 as _distance], metrics=[output_rows=1, elapsed_compute=3.292µs]
+  Take: columns="vector, _rowid, _distance, (id)", metrics=[output_rows=1, elapsed_compute=66.001µs, batches_processed=1, bytes_read=8, iops=1, requests=1]
+   CoalesceBatchesExec: target_batch_size=1024, metrics=[output_rows=1, elapsed_compute=3.333µs]
+    GlobalLimitExec: skip=0, fetch=10, metrics=[output_rows=1, elapsed_compute=167ns]
+     FilterExec: _distance@2 IS NOT NULL, metrics=[output_rows=1, elapsed_compute=8.542µs]
+      SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST], metrics=[output_rows=1, elapsed_compute=63.25µs, row_replacements=1]
+       KNNVectorDistance: metric=l2, metrics=[output_rows=1, elapsed_compute=114.333µs, output_batches=1]
+        LanceScan: uri=/path/to/data, projection=[vector], row_id=true, row_addr=false, ordered=false, metrics=[output_rows=1, elapsed_compute=103.626µs, bytes_read=549, iops=2, requests=2]
+```
+
+#### Inherited from
+
+`StandardQueryBase.analyzePlan`
+
+***
+
+### execute()
+
+```ts
+protected execute(options?): AsyncGenerator<RecordBatch<any>, void, unknown>
+```
+
+Execute the query and return the results as an
+
+#### Parameters
+
+* **options?**: `Partial`&lt;[`QueryExecutionOptions`](../interfaces/QueryExecutionOptions.md)&gt;
+
+#### Returns
+
+`AsyncGenerator`&lt;`RecordBatch`&lt;`any`&gt;, `void`, `unknown`&gt;
+
+#### See
+
+ - AsyncIterator
+of
+ - RecordBatch.
+
+By default, LanceDb will use many threads to calculate results and, when
+the result set is large, multiple batches will be processed at one time.
+This readahead is limited however and backpressure will be applied if this
+stream is consumed slowly (this constrains the maximum memory used by a
+single query)
+
+#### Inherited from
+
+`StandardQueryBase.execute`
+
+***
+
+### explainPlan()
+
+```ts
+explainPlan(verbose): Promise<string>
+```
+
+Generates an explanation of the query execution plan.
+
+#### Parameters
+
+* **verbose**: `boolean` = `false`
+    If true, provides a more detailed explanation. Defaults to false.
+
+#### Returns
+
+`Promise`&lt;`string`&gt;
+
+A Promise that resolves to a string containing the query execution plan explanation.
+
+#### Example
+
+```ts
+import * as lancedb from "@lancedb/lancedb"
+const db = await lancedb.connect("./.lancedb");
+const table = await db.createTable("my_table", [
+  { vector: [1.1, 0.9], id: "1" },
+]);
+const plan = await table.query().nearestTo([0.5, 0.2]).explainPlan();
+```
+
+#### Inherited from
+
+`StandardQueryBase.explainPlan`
+
+***
+
+### fastSearch()
+
+```ts
+fastSearch(): this
+```
+
+Skip searching un-indexed data. This can make search faster, but will miss
+any data that is not yet indexed.
+
+Use [Table#optimize](Table.md#optimize) to index all un-indexed data.
+
+#### Returns
+
+`this`
+
+#### Inherited from
+
+`StandardQueryBase.fastSearch`
+
+***
+
+### ~~filter()~~
+
+```ts
+filter(predicate): this
+```
+
+A filter statement to be applied to this query.
+
+#### Parameters
+
+* **predicate**: `string`
+
+#### Returns
+
+`this`
+
+#### See
+
+where
+
+#### Deprecated
+
+Use `where` instead
+
+#### Inherited from
+
+`StandardQueryBase.filter`
+
+***
+
+### fullTextSearch()
+
+```ts
+fullTextSearch(query, options?): this
+```
+
+#### Parameters
+
+* **query**: `string` \| [`FullTextQuery`](../interfaces/FullTextQuery.md)
+
+* **options?**: `Partial`&lt;[`FullTextSearchOptions`](../interfaces/FullTextSearchOptions.md)&gt;
+
+#### Returns
+
+`this`
+
+#### Inherited from
+
+`StandardQueryBase.fullTextSearch`
+
+***
+
+### limit()
+
+```ts
+limit(limit): this
+```
+
+Set the maximum number of results to return.
+
+By default, a plain search has no limit.  If this method is not
+called then every valid row from the table will be returned.
+
+#### Parameters
+
+* **limit**: `number`
+
+#### Returns
+
+`this`
+
+#### Inherited from
+
+`StandardQueryBase.limit`
+
+***
+
+### offset()
+
+```ts
+offset(offset): this
+```
+
+Set the number of rows to skip before returning results.
+
+This is useful for pagination.
+
+#### Parameters
+
+* **offset**: `number`
+
+#### Returns
+
+`this`
+
+#### Inherited from
+
+`StandardQueryBase.offset`
+
+***
+
+### orderBy()
+
+```ts
+orderBy(ordering): this
+```
+
+Sort the results by the specified column(s).
+
+#### Parameters
+
+* **ordering**: [`ColumnOrdering`](../interfaces/ColumnOrdering.md) \| [`ColumnOrdering`](../interfaces/ColumnOrdering.md)[]
+
+#### Returns
+
+`this`
+
+This query builder.
+
+#### Inherited from
+
+`StandardQueryBase.orderBy`
+
+***
+
+### outputSchema()
+
+```ts
+outputSchema(): Promise<Schema<any>>
+```
+
+Returns the schema of the output that will be returned by this query.
+
+This can be used to inspect the types and names of the columns that will be
+returned by the query before executing it.
+
+#### Returns
+
+`Promise`&lt;`Schema`&lt;`any`&gt;&gt;
+
+An Arrow Schema describing the output columns.
+
+#### Inherited from
+
+`StandardQueryBase.outputSchema`
+
+***
+
+### select()
+
+```ts
+select(columns): this
+```
+
+Return only the specified columns.
+
+By default a query will return all columns from the table.  However, this can have
+a very significant impact on latency.  LanceDb stores data in a columnar fashion.  This
+means we can finely tune our I/O to select exactly the columns we need.
+
+As a best practice you should always limit queries to the columns that you need.  If you
+pass in an array of column names then only those columns will be returned.
+
+You can also use this method to create new "dynamic" columns based on your existing columns.
+For example, you may not care about "a" or "b" but instead simply want "a + b".  This is often
+seen in the SELECT clause of an SQL query (e.g. `SELECT a+b FROM my_table`).
+
+To create dynamic columns you can pass in a Map<string, string>.  A column will be returned
+for each entry in the map.  The key provides the name of the column.  The value is
+an SQL string used to specify how the column is calculated.
+
+For example, an SQL query might state `SELECT a + b AS combined, c`.  The equivalent
+input to this method would be:
+
+#### Parameters
+
+* **columns**: `string` \| `string`[] \| `Record`&lt;`string`, `string`&gt; \| `Map`&lt;`string`, `string`&gt;
+
+#### Returns
+
+`this`
+
+#### Example
+
+```ts
+new Map([["combined", "a + b"], ["c", "c"]])
+
+Columns will always be returned in the order given, even if that order is different than
+the order used when adding the data.
+
+Note that you can pass in a `Record<string, string>` (e.g. an object literal). This method
+uses `Object.entries` which should preserve the insertion order of the object.  However,
+object insertion order is easy to get wrong and `Map` is more foolproof.
+```
+
+#### Inherited from
+
+`StandardQueryBase.select`
+
+***
+
+### toArray()
+
+```ts
+toArray(options?): Promise<any[]>
+```
+
+Collect the results as an array of objects.
+
+#### Parameters
+
+* **options?**: `Partial`&lt;[`QueryExecutionOptions`](../interfaces/QueryExecutionOptions.md)&gt;
+
+#### Returns
+
+`Promise`&lt;`any`[]&gt;
+
+#### Inherited from
+
+`StandardQueryBase.toArray`
+
+***
+
+### toArrow()
+
+```ts
+toArrow(options?): Promise<Table<any>>
+```
+
+Collect the results as an Arrow
+
+#### Parameters
+
+* **options?**: `Partial`&lt;[`QueryExecutionOptions`](../interfaces/QueryExecutionOptions.md)&gt;
+
+#### Returns
+
+`Promise`&lt;`Table`&lt;`any`&gt;&gt;
+
+#### See
+
+ArrowTable.
+
+#### Inherited from
+
+`StandardQueryBase.toArrow`
+
+***
+
+### useLsm()
+
+```ts
+useLsm(enable): this
+```
+
+Control MemWAL read routing for this query.
+
+By default (unset), when the table carries a MemWAL write spec (see
+[Table#setLsmWriteSpec](Table.md#setlsmwritespec)), reads are routed through the LSM scanner so
+they also return data written via the `mergeInsert` LSM path that has not yet
+been compacted into the base table (the active/frozen in-memory memtables and
+the flushed generations), deduplicated by primary key; a table without a spec
+reads the base table.
+
+#### Parameters
+
+* **enable**: `boolean`
+    `true` forces the LSM scanner and errors if the table has no
+    MemWAL write spec. `false` bypasses the MemWAL and reads the base table only,
+    even when a spec is present.
+    Note: the LSM scanner does not support every query shape (e.g. reranking,
+    hybrid search, `orderBy`). On a MemWAL table those shapes error unless
+    `useLsm(false)` is set, because a base-only read would silently exclude
+    un-compacted MemWAL data.
+
+#### Returns
+
+`this`
+
+#### Inherited from
+
+`StandardQueryBase.useLsm`
+
+***
+
+### where()
+
+```ts
+where(predicate): this
+```
+
+A filter statement to be applied to this query.
+
+The filter should be supplied as an SQL query string.  For example:
+
+#### Parameters
+
+* **predicate**: `string`
+
+#### Returns
+
+`this`
+
+#### Example
+
+```ts
+x > 10
+y > 0 AND y < 100
+x > 5 OR y = 'test'
+
+Filtering performance can often be improved by creating a scalar index
+on the filter column(s).
+
+Calling this multiple times combines the filters with a logical AND rather
+than replacing the previous filter.
+```
+
+#### Inherited from
+
+`StandardQueryBase.where`
+
+***
+
+### withRowId()
+
+```ts
+withRowId(): this
+```
+
+Whether to return the row id in the results.
+
+This column can be used to match results between different queries. For
+example, to match results from a full text search and a vector search in
+order to perform hybrid search.
+
+#### Returns
+
+`this`
+
+#### Inherited from
+
+`StandardQueryBase.withRowId`
@@ -942,7 +942,7 @@ Get the schema of the table.
 abstract search(
   query,
   queryType?,
-   ftsColumns?): Query | VectorQuery
+   ftsColumns?): Query | VectorQuery | AutoQuery
 ```

 Create a search query to find the nearest neighbors
@@ -964,7 +964,7 @@ of the given query

 #### Returns

-[`Query`](Query.md) \| [`VectorQuery`](VectorQuery.md)
+[`Query`](Query.md) \| [`VectorQuery`](VectorQuery.md) \| [`AutoQuery`](AutoQuery.md)

 ***

@@ -18,6 +18,7 @@

 ## Classes

+- [AutoQuery](classes/AutoQuery.md)
 - [BooleanQuery](classes/BooleanQuery.md)
 - [BoostQuery](classes/BoostQuery.md)
 - [BranchContents](classes/BranchContents.md)
@@ -10,16 +10,12 @@
 function getRegistry(): EmbeddingFunctionRegistry
 ```

-Utility function to get the global instance of the registry
+Get the global embedding function registry.
+
+LanceDB built-in providers are initialized when this public API is first
+used, so importing the root package does not change automatic search
+selection for tables without embedding metadata.

 ## Returns

 [`EmbeddingFunctionRegistry`](../classes/EmbeddingFunctionRegistry.md)
-
-`EmbeddingFunctionRegistry` The global instance of the registry
-
-## Example
-
-```ts
-const registry = getRegistry();
-const openai = registry.get("openai").create();
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>11.0.0-beta.22</lance-core.version>
+        <lance-core.version>12.0.0-beta.2</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>
@@ -0,0 +1,95 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import { execFileSync } from "node:child_process";
+import { resolve } from "node:path";
+
+import type { OpenAIEmbeddingFunction } from "../lancedb/embedding/openai";
+import type { EmbeddingFunctionRegistry } from "../lancedb/embedding/registry";
+
+type EmbeddingModule = typeof import("../lancedb/embedding");
+type OpenAIModule = typeof import("../lancedb/embedding/openai");
+type RegistryModule = typeof import("../lancedb/embedding/registry");
+
+describe("embedding function registry", () => {
+  const registries: EmbeddingFunctionRegistry[] = [];
+
+  afterEach(() => {
+    for (const registry of registries) {
+      registry.reset();
+    }
+    registries.length = 0;
+  });
+
+  it("defers built-in providers until the public registry API is used", () => {
+    jest.isolateModules(() => {
+      const embedding = require("../lancedb/embedding") as EmbeddingModule;
+      const { getRegistry: getInternalRegistry } =
+        require("../lancedb/embedding/registry") as RegistryModule;
+      const registry = getInternalRegistry();
+      registries.push(registry);
+
+      expect(registry.length()).toBe(0);
+      expect(embedding.getRegistry()).toBe(registry);
+      expect(registry.get("openai")).toBeDefined();
+      expect(registry.get("huggingface")).toBeDefined();
+    });
+  });
+
+  it("preserves automatic FTS search in a fresh process", () => {
+    execFileSync(
+      process.execPath,
+      [resolve(__dirname, "fixtures", "auto_fts_search.cjs")],
+      { stdio: "pipe" },
+    );
+  });
+
+  it("shares registrations across duplicated provider module graphs", () => {
+    let registeringRegistry: EmbeddingFunctionRegistry | undefined;
+    let latestOpenAIConstructor: typeof OpenAIEmbeddingFunction | undefined;
+
+    jest.isolateModules(() => {
+      require("../lancedb/embedding/openai");
+      const { getRegistry } =
+        require("../lancedb/embedding/registry") as RegistryModule;
+      registeringRegistry = getRegistry();
+      registries.push(registeringRegistry);
+      expect(registeringRegistry.get("openai")).toBeDefined();
+    });
+
+    expect(() => {
+      jest.isolateModules(() => {
+        const { OpenAIEmbeddingFunction } =
+          require("../lancedb/embedding/openai") as OpenAIModule;
+        latestOpenAIConstructor = OpenAIEmbeddingFunction;
+        const { getRegistry } =
+          require("../lancedb/embedding/registry") as RegistryModule;
+        registries.push(getRegistry());
+      });
+    }).not.toThrow();
+
+    const previousApiKey = process.env.OPENAI_API_KEY;
+    process.env.OPENAI_API_KEY = "test";
+    try {
+      const latestOpenAI = registeringRegistry!
+        .get<OpenAIEmbeddingFunction>("openai")!
+        .create();
+      expect(latestOpenAI).toBeInstanceOf(latestOpenAIConstructor!);
+    } finally {
+      if (previousApiKey === undefined) {
+        delete process.env.OPENAI_API_KEY;
+      } else {
+        process.env.OPENAI_API_KEY = previousApiKey;
+      }
+    }
+
+    jest.isolateModules(() => {
+      const { getRegistry } =
+        require("../lancedb/embedding") as EmbeddingModule;
+      const publicRegistry = getRegistry();
+      registries.push(publicRegistry);
+      expect(publicRegistry).toBe(registeringRegistry);
+      expect(publicRegistry.get("openai")).toBeDefined();
+    });
+  });
+});
@@ -0,0 +1,33 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+const assert = require("node:assert/strict");
+const tmp = require("tmp");
+const { connect, embedding, Index } = require("../../dist");
+const { getRegistry } = require("../../dist/embedding/registry");
+
+async function main() {
+  assert.equal(typeof embedding.getRegistry, "function");
+  assert.equal(getRegistry().length(), 0);
+  assert.equal(embedding.getRegistry(), getRegistry());
+  assert.equal(getRegistry().length(), 2);
+
+  const dir = tmp.dirSync({ unsafeCleanup: true });
+  let db;
+  try {
+    db = await connect(dir.name);
+    const table = await db.createTable("docs", [{ text: "hello world" }]);
+    await table.createIndex("text", { config: Index.fts() });
+
+    const rows = await table.search("hello").toArray();
+    assert.equal(rows[0].text, "hello world");
+  } finally {
+    db?.close();
+    dir.removeCallback();
+  }
+}
+
+main().catch((error) => {
+  console.error(error);
+  process.exitCode = 1;
+});
@@ -11,10 +11,13 @@ import * as arrow17 from "apache-arrow-17";
 import * as arrow18 from "apache-arrow-18";

 import {
+  AutoQuery,
  Connection,
  MatchQuery,
  PhraseQuery,
+  Query,
  Table,
+  VectorQuery,
  connect,
  tokenize,
 } from "../lancedb";
@@ -682,6 +685,56 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
  },
 );

+// https://github.com/lancedb/lancedb/issues/1963
+it("should query documents with LangChain PDF metadata", async () => {
+  const tmpDir = tmp.dirSync({ unsafeCleanup: true });
+  try {
+    const db = await connect(tmpDir.name);
+    const documents = [
+      {
+        text: "first page",
+        vector: [1, 0],
+        source: "first.pdf",
+        loc: { pageNumber: 1, lines: { from: 1, to: 12 } },
+        pdf: {
+          version: "1.10.100",
+          info: {
+            format: "PDF 1.7",
+            producer: "pdf.js",
+            creator: "Writer",
+          },
+          totalPages: 2,
+        },
+      },
+      {
+        text: "second page",
+        vector: [0, 1],
+        source: "second.pdf",
+        loc: { pageNumber: 2, lines: { from: 13, to: 24 } },
+        pdf: {
+          version: "1.10.100",
+          info: {
+            format: "PDF 1.7",
+            producer: "pdf.js",
+            creator: "Writer",
+          },
+          totalPages: 2,
+        },
+      },
+    ];
+    const documentsTable = await db.createTable("documents", documents);
+
+    const results = await documentsTable.query().toArray();
+
+    expect(results).toHaveLength(2);
+    expect(results[0].source).toBe("first.pdf");
+    expect(results[0].pdf.info.producer).toBe("pdf.js");
+    expect(results[1].loc.pageNumber).toBe(2);
+  } finally {
+    tmpDir.removeCallback();
+  }
+});
+
 describe("merge insert", () => {
  let tmpDir: tmp.DirResult;
  let table: Table;
@@ -1777,6 +1830,194 @@ describe("Read consistency interval", () => {
  });
 });

+describe("automatic search schema consistency", () => {
+  let tmpDir: tmp.DirResult;
+
+  class SchemaRefreshEmbedding extends EmbeddingFunction<string> {
+    ndims() {
+      return 2;
+    }
+
+    embeddingDataType() {
+      return new Float32();
+    }
+
+    async computeSourceEmbeddings(data: string[]) {
+      return data.map((value) => [value.length, 1]);
+    }
+
+    async computeQueryEmbeddings(value: string) {
+      return [value.length, 1];
+    }
+  }
+
+  function embeddingSchema() {
+    const func = new SchemaRefreshEmbedding();
+    return LanceSchema({
+      text: func.sourceField(new Utf8()),
+      vector: func.vectorField(),
+    });
+  }
+
+  beforeEach(() => {
+    getRegistry().reset();
+    register("schema-refresh")(SchemaRefreshEmbedding);
+    tmpDir = tmp.dirSync({ unsafeCleanup: true });
+  });
+
+  afterEach(() => {
+    getRegistry().reset();
+    tmpDir.removeCallback();
+  });
+
+  it("uses the schema refreshed from another connection", async () => {
+    const first = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+    const second = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+
+    try {
+      const stale = await first.createTable("docs", [{ text: "before" }], {
+        schema: embeddingSchema(),
+      });
+      const replacement = await second.createTable(
+        "docs",
+        [{ text: "after hello" }],
+        { mode: "overwrite" },
+      );
+      await replacement.createIndex("text", { config: Index.fts() });
+
+      const search = stale.search("hello");
+      expect(search).toBeInstanceOf(AutoQuery);
+      expect(search).not.toBeInstanceOf(Query);
+      expect(search).not.toBeInstanceOf(VectorQuery);
+      expect("nprobes" in search).toBe(false);
+
+      const rows = await search.toArray();
+      expect(rows[0].text).toBe("after hello");
+      expect((await stale.schema()).metadata.has("embedding_functions")).toBe(
+        false,
+      );
+    } finally {
+      first.close();
+      second.close();
+    }
+  });
+
+  it("tracks embedding metadata across checkout and restore", async () => {
+    const first = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+    const second = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+
+    try {
+      await first.createTable("docs", [{ text: "before" }], {
+        schema: embeddingSchema(),
+      });
+      const table = await second.createTable(
+        "docs",
+        [{ text: "after hello" }],
+        { mode: "overwrite" },
+      );
+      await table.createIndex("text", { config: Index.fts() });
+
+      await table.checkout(1);
+      expect((await table.search("before").toArray())[0].text).toBe("before");
+
+      await table.checkoutLatest();
+      expect((await table.search("hello").toArray())[0].text).toBe(
+        "after hello",
+      );
+
+      await table.checkout(1);
+      await table.restore();
+      expect((await table.search("before").toArray())[0].text).toBe("before");
+    } finally {
+      first.close();
+      second.close();
+    }
+  });
+
+  it("pins automatic search while computing an embedding", async () => {
+    let markStarted!: () => void;
+    let releaseEmbedding!: () => void;
+    const started = new Promise<void>((resolve) => {
+      markStarted = resolve;
+    });
+    const released = new Promise<void>((resolve) => {
+      releaseEmbedding = resolve;
+    });
+
+    class BlockingEmbedding extends SchemaRefreshEmbedding {
+      async computeQueryEmbeddings(value: string) {
+        markStarted();
+        await released;
+        return [value.length, 1];
+      }
+    }
+
+    register("schema-refresh-blocking")(BlockingEmbedding);
+    const func = new BlockingEmbedding();
+    const schema = LanceSchema({
+      text: func.sourceField(new Utf8()),
+      vector: func.vectorField(),
+    });
+    const first = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+    const second = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+
+    try {
+      const table = await first.createTable(
+        "docs",
+        [{ text: "hello before" }],
+        { schema },
+      );
+      const pending = table.search("hello").toArray();
+      await started;
+
+      const replacement = await second.createTable(
+        "docs",
+        [{ text: "hello after" }],
+        { mode: "overwrite" },
+      );
+      await replacement.createIndex("text", { config: Index.fts() });
+      releaseEmbedding();
+
+      expect((await pending)[0].text).toBe("hello before");
+    } finally {
+      releaseEmbedding();
+      first.close();
+      second.close();
+    }
+  });
+
+  it("refreshes a reused automatic search for every execution", async () => {
+    const first = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+    const second = await connect(tmpDir.name, { readConsistencyInterval: 0 });
+
+    try {
+      const table = await first.createTable("docs", [
+        { text: "hello before", marker: "before" },
+      ]);
+      await table.createIndex("text", { config: Index.fts() });
+      const search = table.search("hello").select(["text"]);
+
+      const before = (await search.toArray())[0];
+      expect(before.text).toBe("hello before");
+      expect(before.marker).toBeUndefined();
+
+      const replacement = await second.createTable(
+        "docs",
+        [{ text: "hello after", marker: "after" }],
+        { mode: "overwrite" },
+      );
+      await replacement.createIndex("text", { config: Index.fts() });
+
+      const after = (await search.toArray())[0];
+      expect(after.text).toBe("hello after");
+      expect(after.marker).toBeUndefined();
+    } finally {
+      first.close();
+      second.close();
+    }
+  });
+});
+
 describe("schema evolution", function () {
  let tmpDir: tmp.DirResult;
  beforeEach(() => {
@@ -2344,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] },
@@ -2366,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 = [
@@ -2916,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 = [];
@@ -4,7 +4,15 @@
 import { Field, Schema } from "../arrow";
 import { sanitizeType } from "../sanitize";
 import { EmbeddingFunction } from "./embedding_function";
-import { EmbeddingFunctionConfig, getRegistry } from "./registry";
+import {
+  EmbeddingFunctionConfig,
+  EmbeddingFunctionRegistry,
+  getRegistry as getGlobalRegistry,
+  registerBuiltIn,
+} from "./registry";
+
+type OpenAIModule = typeof import("./openai");
+type TransformersModule = typeof import("./transformers");

 export {
  FieldOptions,
@@ -14,7 +22,39 @@ export {
  EmbeddingFunctionConstructor,
 } from "./embedding_function";

-export * from "./registry";
+export {
+  EmbeddingFunctionRegistry,
+  parseEmbeddingMetadata,
+  register,
+} from "./registry";
+export type {
+  CreateReturnType,
+  EmbeddingFunctionConfig,
+  EmbeddingFunctionCreate,
+  EmbeddingMetadataEntry,
+  ResolvedEmbeddingFunctionConfig,
+} from "./registry";
+
+function initializeBuiltInProviders() {
+  const { OpenAIEmbeddingFunction } = require("./openai") as OpenAIModule;
+  const { TransformersEmbeddingFunction } =
+    require("./transformers") as TransformersModule;
+
+  registerBuiltIn("openai", OpenAIEmbeddingFunction);
+  registerBuiltIn("huggingface", TransformersEmbeddingFunction);
+}
+
+/**
+ * Get the global embedding function registry.
+ *
+ * LanceDB built-in providers are initialized when this public API is first
+ * used, so importing the root package does not change automatic search
+ * selection for tables without embedding metadata.
+ */
+export function getRegistry(): EmbeddingFunctionRegistry {
+  initializeBuiltInProviders();
+  return getGlobalRegistry();
+}

 /**
 * Create a schema with embedding functions.
@@ -5,14 +5,13 @@ import type OpenAI from "openai";
 import type { EmbeddingCreateParams } from "openai/resources/index";
 import { Float, Float32 } from "../arrow";
 import { EmbeddingFunction } from "./embedding_function";
-import { register } from "./registry";
+import { registerBuiltIn } from "./registry";

 export type OpenAIOptions = {
  apiKey: string;
  model: EmbeddingCreateParams["model"];
 };

-@register("openai")
 export class OpenAIEmbeddingFunction extends EmbeddingFunction<
  string,
  Partial<OpenAIOptions>
@@ -100,3 +99,5 @@ export class OpenAIEmbeddingFunction extends EmbeddingFunction<
    return response.data[0].embedding;
  }
 }
+
+registerBuiltIn("openai", OpenAIEmbeddingFunction);
@@ -7,6 +7,10 @@ import {
 } from "./embedding_function";
 import "reflect-metadata";

+const builtInFunctionsKey = Symbol.for(
+  "@lancedb/lancedb::embedding-built-in-functions::v1",
+);
+
 export type CreateReturnType<T> = T extends { init: () => Promise<void> }
  ? Promise<T>
  : T;
@@ -59,6 +63,15 @@ export class EmbeddingFunctionRegistry {
    };
  }

+  /** @ignore */
+  setBuiltIn<
+    T extends EmbeddingFunctionConstructor = EmbeddingFunctionConstructor,
+  >(name: string, ctor: T): T {
+    this.#functions.set(name, ctor);
+    Reflect.defineMetadata("lancedb::embedding::name", name, ctor);
+    return ctor;
+  }
+
  get<T extends EmbeddingFunction<unknown>>(
    name: string,
  ): EmbeddingFunctionCreate<T> | undefined;
@@ -96,6 +109,7 @@ export class EmbeddingFunctionRegistry {
   */
  reset(this: EmbeddingFunctionRegistry) {
    this.#functions.clear();
+    getBuiltInFunctions(this).clear();
  }

  /**
@@ -183,12 +197,56 @@ export class EmbeddingFunctionRegistry {
  }
 }

-const _REGISTRY = new EmbeddingFunctionRegistry();
+function getBuiltInFunctions(registry: EmbeddingFunctionRegistry): Set<string> {
+  const registryWithBuiltIns = registry as EmbeddingFunctionRegistry & {
+    [key: symbol]: Set<string> | undefined;
+  };
+  let builtInFunctions = registryWithBuiltIns[builtInFunctionsKey];
+  if (builtInFunctions === undefined) {
+    builtInFunctions = new Set<string>();
+    registryWithBuiltIns[builtInFunctionsKey] = builtInFunctions;
+  }
+  return builtInFunctions;
+}
+
+// Server bundlers can load the side-effect embedding entry points and the public
+// embedding API from separate module graphs. Keep their registry shared.
+const registryKey = Symbol.for(
+  "@lancedb/lancedb::embedding-function-registry::v1",
+);
+const registryGlobal = globalThis as typeof globalThis & {
+  [key: symbol]: EmbeddingFunctionRegistry | undefined;
+};
+
+function getGlobalRegistry(): EmbeddingFunctionRegistry {
+  const existingRegistry = registryGlobal[registryKey];
+  if (existingRegistry !== undefined) {
+    return existingRegistry;
+  }
+  const registry = new EmbeddingFunctionRegistry();
+  registryGlobal[registryKey] = registry;
+  return registry;
+}
+
+const _REGISTRY = getGlobalRegistry();

 export function register(name?: string) {
  return _REGISTRY.register(name);
 }

+/** @ignore */
+export function registerBuiltIn<
+  T extends EmbeddingFunctionConstructor = EmbeddingFunctionConstructor,
+>(name: string, ctor: T): T {
+  const builtInFunctions = getBuiltInFunctions(_REGISTRY);
+  if (builtInFunctions.has(name)) {
+    return _REGISTRY.setBuiltIn(name, ctor);
+  }
+  _REGISTRY.register(name)(ctor);
+  builtInFunctions.add(name);
+  return ctor;
+}
+
 /**
 * Utility function to get the global instance of the registry
 * @returns `EmbeddingFunctionRegistry` The global instance of the registry
@@ -3,7 +3,7 @@

 import { Float, Float32 } from "../arrow";
 import { EmbeddingFunction } from "./embedding_function";
-import { register } from "./registry";
+import { registerBuiltIn } from "./registry";

 export type XenovaTransformerOptions = {
  /** The wasm compatible model to use */
@@ -31,7 +31,6 @@ export type XenovaTransformerOptions = {
  };
 };

-@register("huggingface")
 export class TransformersEmbeddingFunction extends EmbeddingFunction<
  string,
  Partial<XenovaTransformerOptions>
@@ -158,6 +157,8 @@ export class TransformersEmbeddingFunction extends EmbeddingFunction<
  }
 }

+registerBuiltIn("huggingface", TransformersEmbeddingFunction);
+
 const tensorDiv = (
  src: import("@huggingface/transformers").Tensor,
  divBy: number,
@@ -103,6 +103,7 @@ export {
 } from "./native.js";

 export {
+  AutoQuery,
  ExecutableQuery,
  Query,
  QueryBase,
@@ -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}
@@ -111,13 +134,15 @@ export class QueryBase<
  NativeQueryType extends NativeQuery | NativeVectorQuery | NativeTakeQuery,
 > implements AsyncIterable<RecordBatch>
 {
+  protected inner!: NativeQueryType | Promise<NativeQueryType>;
+
  /**
   * @hidden
   */
-  protected constructor(
-    protected inner: NativeQueryType | Promise<NativeQueryType>,
-  ) {
-    // intentionally empty
+  protected constructor(inner?: NativeQueryType | Promise<NativeQueryType>) {
+    if (inner !== undefined) {
+      this.inner = inner;
+    }
  }

  // call a function on the inner (either a promise or the actual object)
@@ -135,6 +160,15 @@ export class QueryBase<
    }
  }

+  /**
+   * Return the native query used by the next terminal operation.
+   *
+   * @hidden
+   */
+  protected async getInner(): Promise<NativeQueryType> {
+    return this.inner;
+  }
+
  /**
   * Return only the specified columns.
   *
@@ -207,16 +241,11 @@ export class QueryBase<
  /**
   * @hidden
   */
-  protected nativeExecute(
+  protected async nativeExecute(
    options?: Partial<QueryExecutionOptions>,
  ): Promise<NativeBatchIterator> {
-    if (this.inner instanceof Promise) {
-      return this.inner.then((inner) =>
-        inner.execute(options?.maxBatchLength, options?.timeoutMs),
-      );
-    } else {
-      return this.inner.execute(options?.maxBatchLength, options?.timeoutMs);
-    }
+    const inner = await this.getInner();
+    return inner.execute(options?.maxBatchLength, options?.timeoutMs);
  }

  /**
@@ -245,12 +274,7 @@ export class QueryBase<
  /** Collect the results as an Arrow @see {@link ArrowTable}. */
  async toArrow(options?: Partial<QueryExecutionOptions>): Promise<ArrowTable> {
    const batches = [];
-    let inner;
-    if (this.inner instanceof Promise) {
-      inner = await this.inner;
-    } else {
-      inner = this.inner;
-    }
+    const inner = await this.getInner();
    for await (const batch of new RecordBatchIterable(inner, options)) {
      batches.push(batch);
    }
@@ -279,11 +303,8 @@ export class QueryBase<
   * @returns A Promise that resolves to a string containing the query execution plan explanation.
   */
  async explainPlan(verbose = false): Promise<string> {
-    if (this.inner instanceof Promise) {
-      return this.inner.then((inner) => inner.explainPlan(verbose));
-    } else {
-      return this.inner.explainPlan(verbose);
-    }
+    const inner = await this.getInner();
+    return inner.explainPlan(verbose);
  }

  /**
@@ -321,13 +342,8 @@ export class QueryBase<
    distributedMetrics?: AnalyzePlanDistributedMetrics,
  ): Promise<string> {
    const distributedMetricsMode = distributedMetrics ?? "aggregate";
-    if (this.inner instanceof Promise) {
-      return this.inner.then((inner) =>
-        inner.analyzePlan(distributedMetricsMode),
-      );
-    } else {
-      return this.inner.analyzePlan(distributedMetricsMode);
-    }
+    const inner = await this.getInner();
+    return inner.analyzePlan(distributedMetricsMode);
  }

  /**
@@ -339,12 +355,8 @@ export class QueryBase<
   * @returns An Arrow Schema describing the output columns.
   */
  async outputSchema(): Promise<import("./arrow").Schema> {
-    let schemaBuffer: Buffer;
-    if (this.inner instanceof Promise) {
-      schemaBuffer = await this.inner.then((inner) => inner.outputSchema());
-    } else {
-      schemaBuffer = await this.inner.outputSchema();
-    }
+    const inner = await this.getInner();
+    const schemaBuffer = await inner.outputSchema();
    const schema = tableFromIPC(schemaBuffer).schema;
    return schema;
  }
@@ -356,7 +368,7 @@ export class StandardQueryBase<
  extends QueryBase<NativeQueryType>
  implements ExecutableQuery
 {
-  constructor(inner: NativeQueryType | Promise<NativeQueryType>) {
+  constructor(inner?: NativeQueryType | Promise<NativeQueryType>) {
    super(inner);
  }

@@ -510,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)
   *
@@ -537,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;
  }
@@ -551,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;
  }

@@ -565,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;
  }

@@ -578,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;
  }

@@ -592,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;
  }

@@ -606,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;
  }

@@ -627,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;
  }

@@ -661,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;
  }

@@ -686,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;
  }

@@ -700,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;
  }

@@ -716,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);
@@ -763,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.
 *
@@ -788,6 +868,51 @@ export class TakeQuery extends QueryBase<NativeTakeQuery> {
  }
 }

+/**
+ * A builder for automatic string searches.
+ *
+ * Automatic search determines whether to use full-text or vector search from
+ * the table revision selected for each execution. This builder exposes the
+ * common operations supported by both query families.
+ *
+ * @hideconstructor
+ */
+export class AutoQuery extends StandardQueryBase<
+  NativeQuery | NativeVectorQuery
+> {
+  private readonly calls: Array<
+    (inner: NativeQuery | NativeVectorQuery) => void
+  > = [];
+
+  /** @hidden */
+  constructor(
+    private readonly createInner: () => Promise<
+      NativeQuery | NativeVectorQuery
+    >,
+  ) {
+    super();
+  }
+
+  /** @hidden */
+  protected override doCall(
+    fn: (inner: NativeQuery | NativeVectorQuery) => void,
+  ) {
+    this.calls.push(fn);
+  }
+
+  /** @hidden */
+  protected override async getInner(): Promise<
+    NativeQuery | NativeVectorQuery
+  > {
+    const calls = [...this.calls];
+    const inner = await this.createInner();
+    for (const call of calls) {
+      call(inner);
+    }
+    return inner;
+  }
+}
+
 /** A builder for LanceDB queries.
 *
 * @see {@link Table#query}, {@link Table#search}
@@ -840,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 {
@@ -43,10 +43,12 @@ import {
  Table as _NativeTable,
 } from "./native";
 import {
+  AutoQuery,
  FullTextQuery,
  Query,
  TakeQuery,
  VectorQuery,
+  createAutoQuery,
  instanceOfFullTextQuery,
 } from "./query";
 import { sanitizeType } from "./sanitize";
@@ -523,7 +525,7 @@ export abstract class Table {
    query: string | IntoVector | MultiVector | FullTextQuery,
    queryType?: string,
    ftsColumns?: string | string[],
-  ): VectorQuery | Query;
+  ): VectorQuery | Query | AutoQuery;
  /**
   * Search the table with a given query vector.
   *
@@ -975,10 +977,11 @@ export class LocalTable extends Table {
    return this.inner.display();
  }

-  private async getEmbeddingFunctions(): Promise<
-    Map<string, EmbeddingFunctionConfig>
-  > {
-    const schema = await this.schema();
+  private async getEmbeddingFunctions(
+    inner: _NativeTable = this.inner,
+  ): Promise<Map<string, EmbeddingFunctionConfig>> {
+    const schemaBuf = await inner.schema();
+    const schema = tableFromIPC(schemaBuf).schema;
    const registry = getRegistry();
    return registry.parseFunctions(schema.metadata);
  }
@@ -1160,7 +1163,7 @@ export class LocalTable extends Table {
    query: string | IntoVector | MultiVector | FullTextQuery,
    queryType: string = "auto",
    ftsColumns?: string | string[],
-  ): VectorQuery | Query {
+  ): VectorQuery | Query | AutoQuery {
    if (typeof query !== "string" && !instanceOfFullTextQuery(query)) {
      if (queryType === "fts") {
        throw new Error("Cannot perform full text search on a vector query");
@@ -1175,14 +1178,28 @@ export class LocalTable extends Table {
      });
    }

-    // The query type is auto or vector
-    // fall back to full text search if no embedding functions are defined and the query is a string
-    if (
-      queryType === "auto" &&
-      (getRegistry().length() === 0 || instanceOfFullTextQuery(query))
-    ) {
-      return this.query().fullTextSearch(query, {
-        columns: ftsColumns,
+    if (queryType === "auto") {
+      if (instanceOfFullTextQuery(query)) {
+        return this.query().fullTextSearch(query, {
+          columns: ftsColumns,
+        });
+      }
+
+      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;
+        // 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);
      });
    }

@@ -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(
@@ -554,6 +561,12 @@ impl Table {
            .default_error()
    }

+    #[napi(catch_unwind)]
+    pub async fn checkout_current(&self) -> napi::Result<Self> {
+        let table = self.inner_ref()?.checkout_current().await.default_error()?;
+        Ok(Self::new(table))
+    }
+
    #[napi(catch_unwind)]
    pub async fn checkout(&self, version: i64) -> napi::Result<()> {
        self.inner_ref()?
@@ -179,6 +179,18 @@ def connect(
    ...     },
    ... )

+    For Azure Blob Storage, credentials can be passed directly without setting
+    environment variables:
+
+    >>> azure_storage_options = {
+    ...     "account_name": "some-account",
+    ...     "account_key": "some-key",
+    ... }
+    >>> db = lancedb.connect(  # doctest: +SKIP
+    ...     "az://my-container/my-database",
+    ...     storage_options=azure_storage_options,
+    ... )
+
    For tests and temporary data, use an in-memory database:

    >>> db = lancedb.connect("memory://")
@@ -465,6 +477,10 @@ async def connect_async(
    --------

    >>> import lancedb
+    >>> azure_storage_options = {
+    ...     "account_name": "some-account",
+    ...     "account_key": "some-key",
+    ... }
    >>> async def doctest_example():
    ...     # For a local directory, provide a path to the database
    ...     db = await lancedb.connect_async("~/.lancedb")
@@ -472,6 +488,11 @@ async def connect_async(
    ...     db = await lancedb.connect_async("s3://my-bucket/lancedb",
    ...                                      storage_options={
    ...                                          "aws_access_key_id": "***"})
+    ...     # Azure credentials can also be passed directly
+    ...     db = await lancedb.connect_async(
+    ...         "az://my-container/my-database",
+    ...         storage_options=azure_storage_options,
+    ...     )
    ...     # For tests and temporary data, use an in-memory database
    ...     db = await lancedb.connect_async("memory://")
    ...     # Connect to LanceDB cloud
@@ -605,6 +605,7 @@ class FullTextQuery:
 class PyQueryRequest:
    limit: Optional[int]
    offset: Optional[int]
+    take_offsets: Optional[List[int]]
    filter: Optional[Union[str, bytes]]
    full_text_search: Optional[FullTextQuery]
    select: Optional[Union[str, List[str]]]
@@ -109,6 +109,7 @@ def _query_is_plain_scan(query: Query) -> bool:
    return (
        query.vector is None
        and query.full_text_query is None
+        and query.take_offsets is None
        and not query.postfilter
        and not query.order_by
    )
@@ -775,6 +776,10 @@ class Query(pydantic.BaseModel):
    # offset to start fetching results from
    offset: Optional[int] = None

+    # Dataset offsets whose duplicate occurrences must be restored after lookup.
+    # This is populated when a take query is converted to this serializable form.
+    take_offsets: Optional[List[int]] = None
+
    # if true, will only search the indexed data
    fast_search: Optional[bool] = None

@@ -796,6 +801,7 @@ class Query(pydantic.BaseModel):
        query = cls()
        query.limit = req.limit
        query.offset = req.offset
+        query.take_offsets = req.take_offsets
        query.filter = req.filter
        query.full_text_query = req.full_text_search
        query.columns = req.select
@@ -1497,9 +1497,9 @@ class Table(ABC):
        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

-        No guarantees are made regarding the order in which results are returned.  If
-        you desire an output order that matches the order of the given offsets, you will
-        need to add the row offset column to the output and align it yourself.
+        No guarantees are made regarding the order in which results are returned.
+        Repeated offsets produce repeated rows, which makes this method suitable for
+        sampling with replacement.

        Parameters
        ----------
@@ -3883,6 +3883,7 @@ class LanceTable(Table):
            )
            and not self._route_pushdown_to_rust
            and self.current_branch() is None
+            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

@@ -5774,7 +5775,23 @@ class AsyncTable:

    def _sync_query_to_async(
        self, query: Query
-    ) -> AsyncHybridQuery | AsyncFTSQuery | AsyncVectorQuery | AsyncQuery:
+    ) -> (
+        AsyncHybridQuery
+        | AsyncFTSQuery
+        | AsyncVectorQuery
+        | AsyncQuery
+        | AsyncTakeQuery
+    ):
+        if query.take_offsets is not None:
+            take_query = self.take_offsets(query.take_offsets)
+            if query.columns:
+                take_query = take_query.select(query.columns)
+            if query.use_lsm is not None:
+                take_query = take_query.use_lsm(query.use_lsm)
+            if query.with_row_id:
+                take_query = take_query.with_row_id()
+            return take_query
+
        async_query = self.query()
        if query.limit is not None:
            async_query = async_query.limit(query.limit)
@@ -5839,6 +5856,7 @@ class AsyncTable:
                self._namespace_client, self._pushdown_operations
            )
            and not self._route_pushdown_to_rust
+            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

@@ -6329,6 +6347,9 @@ class AsyncTable:
        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

+        No guarantees are made regarding the order in which results are returned.
+        Repeated offsets produce repeated rows.
+
        Parameters
        ----------
        offsets: list[int]
@@ -1891,6 +1891,21 @@ def test_take_queries(tmp_path):
        17,
    ]

+    # Duplicate offsets are occurrences, not set members. Ordering is unspecified.
+    assert sorted(table.take_offsets([5, 2, 5, 17]).to_pandas()["idx"].to_list()) == [
+        2,
+        5,
+        5,
+        17,
+    ]
+
+    # Converting a take builder to its serializable query representation must
+    # retain occurrence metadata and execute with the same multiplicity.
+    query = table.take_offsets([5, 2, 5, 17]).select(["idx"]).to_query_object()
+    assert query.take_offsets == [5, 2, 5, 17]
+    converted = table._execute_query(query).read_all()
+    assert sorted(converted["idx"].to_pylist()) == [2, 5, 5, 17]
+
    # Take by row id
    assert list(
        sorted(table.take_row_ids([5, 2, 17]).to_pandas()["idx"].to_list())
@@ -479,24 +479,49 @@ def test_remote_permutation_is_picklable():
                match = re.search(
                    r"_rowoffset\s+in\s+\((.*?)\)", body["filter"], re.IGNORECASE
                )
-                offsets = [int(o.strip()) for o in match.group(1).split(",")]
+                offsets = list(
+                    dict.fromkeys(int(o.strip()) for o in match.group(1).split(","))
+                )
            else:
                offsets = list(range(len(rows)))
-            table = pa.table({"a": [rows[offset] for offset in offsets]})
+            columns = body.get("columns") or ["a"]
+            table = pa.table(
+                {
+                    column: (
+                        [rows[offset] for offset in offsets]
+                        if column == "a"
+                        else offsets
+                    )
+                    for column in columns
+                }
+            )

            request.send_response(200)
            request.send_header("Content-Type", "application/vnd.apache.arrow.file")
            request.end_headers()
            with pa.ipc.new_file(request.wfile, schema=table.schema) as writer:
-                writer.write_table(table)
+                writer.write_table(table, max_chunksize=2)
        else:
            request.send_response(404)
            request.end_headers()

    with mock_lancedb_connection(handler) as db:
-        permutation = Permutation.identity(db.open_table("test"))
+        table = db.open_table("test")
+        assert table.take_offsets([0, 2, 0, 4]).to_list() == [
+            {"a": 0},
+            {"a": 0},
+            {"a": 2},
+            {"a": 4},
+        ]
+
+        permutation = Permutation.identity(table)
        restored = pickle.loads(pickle.dumps(permutation))
-        assert restored.__getitems__([0, 2, 4]) == [{"a": 0}, {"a": 2}, {"a": 4}]
+        assert restored.__getitems__([0, 2, 0, 4]) == [
+            {"a": 0},
+            {"a": 2},
+            {"a": 0},
+            {"a": 4},
+        ]


 def test_create_table_exist_ok():
@@ -289,6 +289,7 @@ impl<'py> IntoPyObject<'py> for PyQueryVectors {
 pub struct PyQueryRequest {
    pub limit: Option<usize>,
    pub offset: Option<usize>,
+    pub take_offsets: Option<Vec<u64>>,
    pub filter: Option<PyQueryFilter>,
    pub full_text_search: Option<PyLanceDB<FtsQuery>>,
    pub select: PySelect,
@@ -318,6 +319,7 @@ impl From<AnyQuery> for PyQueryRequest {
            AnyQuery::Query(query_request) => Self {
                limit: query_request.limit,
                offset: query_request.offset,
+                take_offsets: query_request.take_offsets,
                filter: query_request.filter.map(PyQueryFilter),
                full_text_search: query_request
                    .full_text_search
@@ -345,6 +347,7 @@ impl From<AnyQuery> for PyQueryRequest {
            AnyQuery::VectorQuery(vector_query) => Self {
                limit: vector_query.base.limit,
                offset: vector_query.base.offset,
+                take_offsets: vector_query.base.take_offsets,
                filter: vector_query.base.filter.map(PyQueryFilter),
                full_text_search: None,
                select: PySelect(vector_query.base.select),
@@ -31,7 +31,7 @@ use lance::io::RecordBatchStream;
 use lance_arrow::RecordBatchExt;
 use lance_core::ROW_ID;
 use lance_core::error::LanceOptionExt;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;

 /// Reads a permutation of a source table based on row IDs stored in a separate table
@@ -234,7 +234,14 @@ impl PermutationReader {
            .expect_ok()?
            .values();

-        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();
+        let mut unique_row_ids = HashSet::with_capacity(num_rows);
+        let in_list: Vec<Expr> = row_ids
+            .iter()
+            .copied()
+            .filter(|row_id| unique_row_ids.insert(*row_id))
+            .map(lit)
+            .collect();
+        let num_unique_row_ids = unique_row_ids.len();

        let base_query = QueryRequest {
            filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
@@ -247,7 +254,7 @@ impl PermutationReader {
            .query(
                &AnyQuery::Query(base_query),
                QueryExecutionOptions {
-                    max_batch_length: num_rows as u32,
+                    max_batch_length: num_unique_row_ids as u32,
                    ..Default::default()
                },
            )
@@ -262,9 +269,9 @@ impl PermutationReader {
            });
        }

-        if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_rows {
+        if batches.iter().map(|b| b.num_rows()).sum::<usize>() != num_unique_row_ids {
            return Err(Error::InvalidInput {
-                message: "Base table returned different number of rows than the number of row IDs"
+                message: "Base table returned a different number of rows than the number of unique row IDs"
                    .to_string(),
            });
        }
@@ -504,6 +511,7 @@ impl PermutationReader {
            let table = Table::from(self.base_table.clone());
            let batches = table
                .take_offsets(offsets.to_vec())
+                .preserve_order()
                .select(selection.clone())
                .execute()
                .await?
@@ -803,10 +811,10 @@ mod tests {
        .unwrap();

        // Take offsets in reverse order and verify returned rows match that order
-        let offsets = vec![5, 3, 1, 0];
+        let offsets = vec![5, 3, 5, 1, 0];
        let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();

-        assert_eq!(batch.num_rows(), 4);
+        assert_eq!(batch.num_rows(), 5);

        let idx_values = batch
            .column(0)
@@ -820,6 +828,52 @@ mod tests {
        assert_eq!(idx_values, expected);
    }

+    #[tokio::test]
+    async fn test_take_offsets_preserves_repeated_rows_in_permutation() {
+        let base_table = lance_datagen::gen_batch()
+            .col("idx", lance_datagen::array::step::<Int32Type>())
+            .into_mem_table("tbl", RowCount::from(5), BatchCount::from(1))
+            .await;
+        let base_row_ids = collect_column::<UInt64Type>(&base_table, "_rowid").await;
+        let permutation_row_ids = vec![
+            base_row_ids[3],
+            base_row_ids[1],
+            base_row_ids[3],
+            base_row_ids[2],
+        ];
+        let permutation_batch = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![
+                Field::new("row_id", DataType::UInt64, false),
+                Field::new(SPLIT_ID_COLUMN, DataType::UInt64, false),
+            ])),
+            vec![
+                Arc::new(UInt64Array::from(permutation_row_ids)),
+                Arc::new(UInt64Array::from(vec![0; 4])),
+            ],
+        )
+        .unwrap();
+        let permutation_table = virtual_table("row_ids", &permutation_batch).await;
+        let reader = PermutationReader::try_from_tables(
+            base_table.base_table().clone(),
+            permutation_table.base_table().clone(),
+            0,
+        )
+        .await
+        .unwrap();
+
+        let batch = reader
+            .take_offsets(&[0, 1, 2, 3], Select::All)
+            .await
+            .unwrap();
+        let idx_values = batch
+            .column(0)
+            .as_primitive::<Int32Type>()
+            .values()
+            .to_vec();
+
+        assert_eq!(idx_values, vec![3, 1, 3, 2]);
+    }
+
    #[tokio::test]
    async fn test_take_offsets_with_column_selection() {
        let (base_table, row_ids_table, row_ids) = setup_permutation_tables(10).await;
@@ -883,17 +937,17 @@ mod tests {
            .unwrap();

        // With no permutation table, take_offsets uses the base table directly
-        let offsets = vec![0, 2, 4, 6];
+        let offsets = vec![0, 2, 0, 4, 6];
        let batch = reader.take_offsets(&offsets, Select::All).await.unwrap();

-        assert_eq!(batch.num_rows(), 4);
+        assert_eq!(batch.num_rows(), 5);

        let idx_values = batch
            .column(0)
            .as_primitive::<Int32Type>()
            .values()
            .to_vec();
-        assert_eq!(idx_values, vec![0, 2, 4, 6]);
+        assert_eq!(idx_values, vec![0, 2, 0, 4, 6]);
    }

    #[tokio::test]
@@ -10,7 +10,7 @@ use lance::io::WrappingObjectStore;
 use object_store::{
    CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
    ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
-    UploadPart, path::Path,
+    UploadPart, list::PaginatedListStore, path::Path,
 };

 use async_trait::async_trait;
@@ -187,6 +187,14 @@ impl WrappingObjectStore for MirroringObjectStoreWrapper {
            secondary: self.secondary.clone(),
        })
    }
+
+    fn wrap_paginated(
+        &self,
+        _store_prefix: &str,
+        original: Arc<dyn PaginatedListStore>,
+    ) -> Option<Arc<dyn PaginatedListStore>> {
+        Some(original)
+    }
 }

 // windows pathing can't be simply concatenated
@@ -12,7 +12,7 @@ use lance::io::WrappingObjectStore;
 use object_store::{
    CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
    PutMultipartOptions, PutOptions, PutPayload, PutResult, RenameOptions, Result as OSResult,
-    UploadPart, path::Path,
+    UploadPart, list::PaginatedListStore, path::Path,
 };

 #[derive(Debug, Default)]
@@ -57,6 +57,14 @@ impl WrappingObjectStore for IoStatsHolder {
            stats: self.0.clone(),
        })
    }
+
+    fn wrap_paginated(
+        &self,
+        _store_prefix: &str,
+        original: Arc<dyn PaginatedListStore>,
+    ) -> Option<Arc<dyn PaginatedListStore>> {
+        Some(original)
+    }
 }

 impl IoTrackingStore {
@@ -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 {
@@ -1,21 +1,37 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

+use std::collections::{HashMap, HashSet};
+use std::pin::Pin;
 use std::sync::Arc;
 use std::{future::Future, time::Duration};

 use arrow::compute::concat_batches;
-use arrow_array::{Array, Float16Array, Float32Array, Float64Array, RecordBatch, make_array};
+use arrow_array::{
+    Array, Float16Array, Float32Array, Float64Array, RecordBatch, UInt64Array,
+    cast::AsArray,
+    make_array,
+    types::{Int64Type, UInt64Type},
+};
 use arrow_schema::{DataType, SchemaRef};
+use datafusion_common::{DataFusionError, Result as DataFusionResult};
+use datafusion_execution::TaskContext;
 use datafusion_expr::{Expr, col, lit};
-use datafusion_physical_plan::ExecutionPlan;
-use futures::{FutureExt, TryFutureExt, TryStreamExt, stream, try_join};
+use datafusion_physical_expr::{EquivalenceProperties, Partitioning};
+use datafusion_physical_plan::{
+    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
+    coalesce_partitions::CoalescePartitionsExec,
+    execution_plan::{Boundedness, EmissionType},
+    limit::GlobalLimitExec,
+    stream::RecordBatchStreamAdapter,
+};
+use futures::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, stream, try_join};
 use half::f16;
 /// Re-export Lance ColumnOrdering type for use in query ordering
 pub use lance::dataset::scanner::ColumnOrdering;
 use lance::dataset::{ROW_ID, scanner::DatasetRecordBatchStream};
 use lance_arrow::RecordBatchExt;
-use lance_datafusion::exec::execute_plan;
+use lance_datafusion::exec::{execute_plan, format_plan as format_analyzed_plan};
 use lance_index::scalar::FullTextSearchQuery;
 use lance_index::scalar::inverted::SCORE_COL;
 use lance_index::vector::DIST_COL;
@@ -822,6 +838,14 @@ pub struct QueryRequest {
    /// Offset of the query.
    pub offset: Option<usize>,

+    /// Dataset offsets whose occurrence multiplicity must be restored after
+    /// executing the physical lookup represented by this request.
+    ///
+    /// This is client-side execution metadata used when a [`TakeQuery`] is
+    /// converted into a request. It is not sent to remote services.
+    #[doc(hidden)]
+    pub take_offsets: Option<Vec<u64>>,
+
    /// Apply filter to the returned rows.
    pub filter: Option<QueryFilter>,

@@ -890,6 +914,7 @@ impl Default for QueryRequest {
        Self {
            limit: None,
            offset: None,
+            take_offsets: None,
            filter: None,
            filter_error: None,
            full_text_search: None,
@@ -1515,6 +1540,302 @@ impl HasQuery for VectorQuery {
    }
 }

+fn take_occurrences(offsets: &[u64]) -> HashMap<u64, usize> {
+    let mut occurrences = HashMap::with_capacity(offsets.len());
+    for offset in offsets {
+        *occurrences.entry(*offset).or_insert(0) += 1;
+    }
+    occurrences
+}
+
+fn restore_take_batch_with_occurrences(
+    batch: RecordBatch,
+    offsets: &[u64],
+    occurrences: &HashMap<u64, usize>,
+    ordering_column: &str,
+    drop_ordering_column: bool,
+    preserve_order: bool,
+) -> Result<RecordBatch> {
+    let actual_offsets = batch
+        .column_by_name(ordering_column)
+        .ok_or_else(|| Error::Schema {
+            message: format!(
+                "take query result did not include ordering column '{ordering_column}'"
+            ),
+        })?;
+    let actual_offsets = match actual_offsets.data_type() {
+        DataType::UInt64 => actual_offsets
+            .as_primitive::<UInt64Type>()
+            .values()
+            .to_vec(),
+        DataType::Int64 => actual_offsets
+            .as_primitive::<Int64Type>()
+            .values()
+            .iter()
+            .map(|offset| {
+                u64::try_from(*offset).map_err(|_| Error::Schema {
+                    message: format!(
+                        "take query ordering column '{ordering_column}' contained a negative offset"
+                    ),
+                })
+            })
+            .collect::<Result<Vec<_>>>()?,
+        data_type => {
+            return Err(Error::Schema {
+                message: format!(
+                    "take query ordering column '{ordering_column}' had unsupported type {data_type}"
+                ),
+            });
+        }
+    };
+
+    let mut desired_order = Vec::with_capacity(offsets.len());
+    if preserve_order {
+        let ordering = actual_offsets
+            .iter()
+            .copied()
+            .enumerate()
+            .map(|(index, offset)| (offset, index as u64))
+            .collect::<HashMap<_, _>>();
+        // Missing offsets retain the filter-based behavior of returning no row.
+        desired_order.extend(
+            offsets
+                .iter()
+                .filter_map(|offset| ordering.get(offset).copied()),
+        );
+    } else {
+        // Public take queries do not guarantee output order. Preserve the lookup's
+        // existing order and only restore the multiplicity of each matching row.
+        for (index, offset) in actual_offsets.iter().enumerate() {
+            if let Some(count) = occurrences.get(offset) {
+                desired_order.extend(std::iter::repeat_n(index as u64, *count));
+            }
+        }
+    }
+
+    let mut ordered_batch = if desired_order.len() == batch.num_rows()
+        && desired_order
+            .iter()
+            .enumerate()
+            .all(|(index, desired)| *desired == index as u64)
+    {
+        batch
+    } else {
+        arrow_select::take::take_record_batch(&batch, &UInt64Array::from(desired_order))?
+    };
+
+    if drop_ordering_column {
+        ordered_batch = ordered_batch.drop_column(ordering_column)?;
+    }
+
+    Ok(ordered_batch)
+}
+
+#[cfg(test)]
+fn restore_take_batch(
+    batch: RecordBatch,
+    offsets: &[u64],
+    ordering_column: &str,
+    drop_ordering_column: bool,
+    preserve_order: bool,
+) -> Result<RecordBatch> {
+    restore_take_batch_with_occurrences(
+        batch,
+        offsets,
+        &take_occurrences(offsets),
+        ordering_column,
+        drop_ordering_column,
+        preserve_order,
+    )
+}
+
+/// Restores the logical offset occurrence sequence above the physical lookup plan.
+///
+/// The lookup plan returns each matching row at most once. For ordinary unordered
+/// takes this operator expands each input batch incrementally and preserves the
+/// lookup's partitioning. The explicitly ordered reader path collects one coalesced
+/// input before restoring requested order. Pagination must remain above this operator
+/// so it applies to occurrences.
+#[derive(Debug)]
+struct TakeRestoreExec {
+    input: Arc<dyn ExecutionPlan>,
+    offsets: Vec<u64>,
+    occurrences: Arc<HashMap<u64, usize>>,
+    ordering_column: String,
+    drop_ordering_column: bool,
+    preserve_order: bool,
+    schema: SchemaRef,
+    properties: Arc<PlanProperties>,
+}
+
+impl TakeRestoreExec {
+    fn try_new(
+        input: Arc<dyn ExecutionPlan>,
+        offsets: Vec<u64>,
+        ordering_column: String,
+        drop_ordering_column: bool,
+        preserve_order: bool,
+    ) -> Result<Self> {
+        let schema = if drop_ordering_column {
+            RecordBatch::new_empty(input.schema())
+                .drop_column(&ordering_column)?
+                .schema()
+        } else {
+            input.schema()
+        };
+        let partition_count = if preserve_order {
+            1
+        } else {
+            input.output_partitioning().partition_count()
+        };
+        let emission_type = if preserve_order {
+            EmissionType::Final
+        } else {
+            EmissionType::Incremental
+        };
+        let properties = Arc::new(PlanProperties::new(
+            EquivalenceProperties::new(schema.clone()),
+            Partitioning::UnknownPartitioning(partition_count),
+            emission_type,
+            Boundedness::Bounded,
+        ));
+
+        Ok(Self {
+            input,
+            occurrences: Arc::new(take_occurrences(&offsets)),
+            offsets,
+            ordering_column,
+            drop_ordering_column,
+            preserve_order,
+            schema,
+            properties,
+        })
+    }
+}
+
+impl DisplayAs for TakeRestoreExec {
+    fn fmt_as(
+        &self,
+        _display_type: DisplayFormatType,
+        formatter: &mut std::fmt::Formatter<'_>,
+    ) -> std::fmt::Result {
+        write!(
+            formatter,
+            "TakeRestoreExec: occurrences={}",
+            self.offsets.len()
+        )
+    }
+}
+
+impl ExecutionPlan for TakeRestoreExec {
+    fn name(&self) -> &str {
+        "TakeRestoreExec"
+    }
+
+    fn properties(&self) -> &Arc<PlanProperties> {
+        &self.properties
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        vec![&self.input]
+    }
+
+    fn maintains_input_order(&self) -> Vec<bool> {
+        vec![!self.preserve_order]
+    }
+
+    fn benefits_from_input_partitioning(&self) -> Vec<bool> {
+        vec![false]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        if children.len() != 1 {
+            return Err(DataFusionError::Internal(format!(
+                "TakeRestoreExec expected one child, got {}",
+                children.len()
+            )));
+        }
+        let child = children.into_iter().next().unwrap();
+        let plan = Self::try_new(
+            child,
+            self.offsets.clone(),
+            self.ordering_column.clone(),
+            self.drop_ordering_column,
+            self.preserve_order,
+        )
+        .map_err(|error| DataFusionError::External(Box::new(error)))?;
+        Ok(Arc::new(plan))
+    }
+
+    fn execute(
+        &self,
+        partition: usize,
+        context: Arc<TaskContext>,
+    ) -> DataFusionResult<datafusion_physical_plan::SendableRecordBatchStream> {
+        let partition_count = self.input.output_partitioning().partition_count();
+        if partition >= partition_count || (self.preserve_order && partition != 0) {
+            return Err(DataFusionError::Internal(format!(
+                "TakeRestoreExec cannot execute partition {partition}; input has {partition_count} partitions"
+            )));
+        }
+
+        let input = self.input.execute(partition, context)?;
+        let output_schema = self.schema.clone();
+        let offsets = self.offsets.clone();
+        let occurrences = self.occurrences.clone();
+        let ordering_column = self.ordering_column.clone();
+        let drop_ordering_column = self.drop_ordering_column;
+        let preserve_order = self.preserve_order;
+        let stream: Pin<Box<dyn futures::Stream<Item = DataFusionResult<RecordBatch>> + Send>> =
+            if preserve_order {
+                let input_schema = input.schema();
+                Box::pin(stream::once(async move {
+                    let batches = input.try_collect::<Vec<_>>().await?;
+                    let batch = if batches.is_empty() {
+                        RecordBatch::new_empty(input_schema.clone())
+                    } else {
+                        concat_batches(&input_schema, &batches)?
+                    };
+                    restore_take_batch_with_occurrences(
+                        batch,
+                        &offsets,
+                        &occurrences,
+                        &ordering_column,
+                        drop_ordering_column,
+                        true,
+                    )
+                    .map_err(|error| DataFusionError::External(Box::new(error)))
+                }))
+            } else {
+                Box::pin(input.map(move |batch| {
+                    batch.and_then(|batch| {
+                        restore_take_batch_with_occurrences(
+                            batch,
+                            &offsets,
+                            &occurrences,
+                            &ordering_column,
+                            drop_ordering_column,
+                            false,
+                        )
+                        .map_err(|error| DataFusionError::External(Box::new(error)))
+                    })
+                }))
+            };
+
+        Ok(Box::pin(RecordBatchStreamAdapter::new(
+            output_schema,
+            stream,
+        )))
+    }
+
+    fn supports_limit_pushdown(&self) -> bool {
+        false
+    }
+}
+
 /// A builder for LanceDB take queries.
 ///
 /// See [`crate::Table::query`] for more details on queries
@@ -1531,6 +1852,8 @@ impl HasQuery for VectorQuery {
 pub struct TakeQuery {
    parent: Arc<dyn BaseTable>,
    request: QueryRequest,
+    offsets: Option<Vec<u64>>,
+    preserve_order: bool,
 }

 impl TakeQuery {
@@ -1538,15 +1861,24 @@ impl TakeQuery {
    ///
    /// See [`crate::Table::take_offsets`] for more details.
    pub fn from_offsets(parent: Arc<dyn BaseTable>, offsets: Vec<u64>) -> Self {
-        let in_list: Vec<Expr> = offsets.iter().map(|o| lit(*o)).collect();
+        let mut seen = HashSet::with_capacity(offsets.len());
+        let in_list: Vec<Expr> = offsets
+            .iter()
+            .copied()
+            .filter(|offset| seen.insert(*offset))
+            .map(lit)
+            .collect();
        Self {
            parent,
            request: QueryRequest {
                filter: Some(QueryFilter::Datafusion(
                    col("_rowoffset").in_list(in_list, false),
                )),
+                take_offsets: Some(offsets.clone()),
                ..Default::default()
            },
+            offsets: Some(offsets),
+            preserve_order: false,
        }
    }

@@ -1561,9 +1893,181 @@ impl TakeQuery {
                filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
                ..Default::default()
            },
+            offsets: None,
+            preserve_order: false,
        }
    }

+    /// Preserve the requested offset order when restoring duplicate occurrences.
+    ///
+    /// This is reserved for readers whose API explicitly guarantees ordering.
+    pub(crate) fn preserve_order(mut self) -> Self {
+        debug_assert!(self.offsets.is_some());
+        self.preserve_order = true;
+        self
+    }
+
+    async fn request_with_row_offset(
+        parent: &dyn BaseTable,
+        request: &QueryRequest,
+    ) -> Result<(QueryRequest, String, bool)> {
+        const ROW_OFFSET: &str = "_rowoffset";
+        const INTERNAL_ROW_OFFSET: &str = "__lancedb_take_row_offset";
+
+        let mut request = request.clone();
+        // The physical lookup must not recursively restore occurrences. The
+        // wrapper above this request owns that logical operation.
+        request.take_offsets = None;
+        let (ordering_column, drop_ordering_column) = match &mut request.select {
+            Select::All => {
+                let mut columns = parent
+                    .schema()
+                    .await?
+                    .fields()
+                    .iter()
+                    .map(|field| field.name().clone())
+                    .collect::<Vec<_>>();
+                columns.push(ROW_OFFSET.to_string());
+                request.select = Select::Columns(columns);
+                (ROW_OFFSET.to_string(), true)
+            }
+            Select::Columns(columns) => {
+                if columns.iter().any(|column| column == ROW_OFFSET) {
+                    (ROW_OFFSET.to_string(), false)
+                } else {
+                    columns.push(ROW_OFFSET.to_string());
+                    (ROW_OFFSET.to_string(), true)
+                }
+            }
+            Select::Dynamic(columns) => {
+                let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
+                while columns.iter().any(|(name, _)| name == &ordering_column) {
+                    ordering_column.push('_');
+                }
+                columns.push((ordering_column.clone(), ROW_OFFSET.to_string()));
+                (ordering_column, true)
+            }
+            Select::Expr(columns) => {
+                let mut ordering_column = INTERNAL_ROW_OFFSET.to_string();
+                while columns.iter().any(|(name, _)| name == &ordering_column) {
+                    ordering_column.push('_');
+                }
+                columns.push((ordering_column.clone(), col(ROW_OFFSET)));
+                (ordering_column, true)
+            }
+        };
+
+        Ok((request, ordering_column, drop_ordering_column))
+    }
+
+    async fn prepare_offsets_lookup(
+        parent: &dyn BaseTable,
+        request: &QueryRequest,
+    ) -> Result<(QueryRequest, String, bool, usize, Option<usize>)> {
+        let (mut request, ordering_column, drop_ordering_column) =
+            Self::request_with_row_offset(parent, request).await?;
+        // The lookup operates on distinct physical rows. Pagination is a logical
+        // operation over occurrences and must be applied only after restoration.
+        let output_offset = request.offset.take().unwrap_or_default();
+        let output_limit = request.limit.take();
+
+        Ok((
+            request,
+            ordering_column,
+            drop_ordering_column,
+            output_offset,
+            output_limit,
+        ))
+    }
+
+    fn wrap_offsets_plan(
+        lookup: Arc<dyn ExecutionPlan>,
+        offsets: &[u64],
+        ordering_column: String,
+        drop_ordering_column: bool,
+        output_offset: usize,
+        output_limit: Option<usize>,
+        preserve_order: bool,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let lookup = if preserve_order {
+            Arc::new(CoalescePartitionsExec::new(lookup)) as Arc<dyn ExecutionPlan>
+        } else {
+            lookup
+        };
+        let restored: Arc<dyn ExecutionPlan> = Arc::new(TakeRestoreExec::try_new(
+            lookup,
+            offsets.to_vec(),
+            ordering_column,
+            drop_ordering_column,
+            preserve_order,
+        )?);
+
+        if output_offset > 0 || output_limit.is_some() {
+            Ok(Arc::new(GlobalLimitExec::new(
+                restored,
+                output_offset,
+                output_limit,
+            )))
+        } else {
+            Ok(restored)
+        }
+    }
+
+    fn wrap_offsets_explanation(
+        lookup: &str,
+        occurrence_count: usize,
+        output_offset: usize,
+        output_limit: Option<usize>,
+        preserve_order: bool,
+    ) -> String {
+        fn indent(plan: &str, spaces: usize) -> String {
+            let indentation = " ".repeat(spaces);
+            plan.lines()
+                .map(|line| format!("{indentation}{line}"))
+                .collect::<Vec<_>>()
+                .join("\n")
+        }
+
+        let restored = if preserve_order {
+            format!(
+                "TakeRestoreExec: occurrences={occurrence_count}\n  CoalescePartitionsExec\n{}",
+                indent(lookup, 4)
+            )
+        } else {
+            format!(
+                "TakeRestoreExec: occurrences={occurrence_count}\n{}",
+                indent(lookup, 2)
+            )
+        };
+
+        if output_offset > 0 || output_limit.is_some() {
+            let fetch = output_limit
+                .map(|limit| limit.to_string())
+                .unwrap_or_else(|| "None".to_string());
+            format!(
+                "GlobalLimitExec: skip={output_offset}, fetch={fetch}\n{}",
+                indent(&restored, 2)
+            )
+        } else {
+            restored
+        }
+    }
+
+    async fn create_offsets_plan(
+        &self,
+        offsets: &[u64],
+        options: QueryExecutionOptions,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        create_take_offsets_plan(
+            self.parent.as_ref(),
+            &self.request,
+            offsets,
+            options,
+            self.preserve_order,
+        )
+        .await
+    }
+
    /// Convert the `TakeQuery` into a `QueryRequest`.
    pub fn into_request(self) -> QueryRequest {
        self.request
@@ -1608,6 +2112,63 @@ impl TakeQuery {
    }
 }

+pub(crate) async fn create_take_offsets_plan(
+    parent: &dyn BaseTable,
+    request: &QueryRequest,
+    offsets: &[u64],
+    options: QueryExecutionOptions,
+    preserve_order: bool,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    let (request, ordering_column, drop_ordering_column, output_offset, output_limit) =
+        TakeQuery::prepare_offsets_lookup(parent, request).await?;
+    let lookup_options = if preserve_order {
+        options.without_output_batch_length_limit()
+    } else {
+        options
+    };
+    let lookup = parent
+        .create_plan(&AnyQuery::Query(request), lookup_options)
+        .await?;
+
+    TakeQuery::wrap_offsets_plan(
+        lookup,
+        offsets,
+        ordering_column,
+        drop_ordering_column,
+        output_offset,
+        output_limit,
+        preserve_order,
+    )
+}
+
+pub(crate) async fn explain_take_offsets_plan(
+    parent: &dyn BaseTable,
+    request: &QueryRequest,
+    offsets: &[u64],
+    verbose: bool,
+) -> Result<String> {
+    let (request, _, _, output_offset, output_limit) =
+        TakeQuery::prepare_offsets_lookup(parent, request).await?;
+    let lookup = parent
+        .explain_plan(&AnyQuery::Query(request), verbose)
+        .await?;
+    Ok(TakeQuery::wrap_offsets_explanation(
+        &lookup,
+        offsets.len(),
+        output_offset,
+        output_limit,
+        false,
+    ))
+}
+
+pub(crate) async fn prepare_take_offsets_request(
+    parent: &dyn BaseTable,
+    request: &QueryRequest,
+) -> Result<QueryRequest> {
+    let (request, _, _, _, _) = TakeQuery::prepare_offsets_lookup(parent, request).await?;
+    Ok(request)
+}
+
 impl HasQuery for TakeQuery {
    fn mut_query(&mut self) -> &mut QueryRequest {
        &mut self.request
@@ -1616,6 +2177,10 @@ impl HasQuery for TakeQuery {

 impl ExecutableQuery for TakeQuery {
    async fn create_plan(&self, options: QueryExecutionOptions) -> Result<Arc<dyn ExecutionPlan>> {
+        if let Some(offsets) = &self.offsets {
+            return self.create_offsets_plan(offsets, options).await;
+        }
+
        let req = AnyQuery::Query(self.request.clone());
        self.parent.clone().create_plan(&req, options).await
    }
@@ -1624,6 +2189,18 @@ impl ExecutableQuery for TakeQuery {
        &self,
        options: QueryExecutionOptions,
    ) -> Result<SendableRecordBatchStream> {
+        if self.offsets.is_some() {
+            let plan = self.create_plan(options.clone()).await?;
+            let inner = execute_plan(plan, Default::default())?;
+            let inner = MaxBatchLengthStream::new_boxed(inner, options.max_batch_length as usize);
+            let inner = if let Some(timeout) = options.timeout {
+                TimeoutStream::new_boxed(inner, timeout)
+            } else {
+                inner
+            };
+            return Ok(DatasetRecordBatchStream::new(inner).into());
+        }
+
        let query = AnyQuery::Query(self.request.clone());
        Ok(SendableRecordBatchStream::from(
            self.parent.clone().query(&query, options).await?,
@@ -1631,11 +2208,51 @@ impl ExecutableQuery for TakeQuery {
    }

    async fn explain_plan(&self, verbose: bool) -> Result<String> {
+        if let Some(offsets) = &self.offsets {
+            let (request, _, _, output_offset, output_limit) =
+                Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
+            // Ask the backend to explain only the distinct-row lookup. This keeps
+            // remote explanation non-executing while still showing the client-side
+            // operators that create_plan and execution place above that lookup.
+            let lookup = self
+                .parent
+                .explain_plan(&AnyQuery::Query(request), verbose)
+                .await?;
+            return Ok(Self::wrap_offsets_explanation(
+                &lookup,
+                offsets.len(),
+                output_offset,
+                output_limit,
+                self.preserve_order,
+            ));
+        }
+
        let query = AnyQuery::Query(self.request.clone());
        self.parent.explain_plan(&query, verbose).await
    }

    async fn analyze_plan_with_options(&self, options: QueryExecutionOptions) -> Result<String> {
+        if self.offsets.is_some() {
+            if self.parent.analyze_plan_is_remote() {
+                let (request, _, _, _, _) =
+                    Self::prepare_offsets_lookup(self.parent.as_ref(), &self.request).await?;
+                // Remote analysis is owned by the service. The current wire
+                // request represents only the distinct-row lookup, so return
+                // the service report unchanged instead of fabricating metrics
+                // for client-side restoration operators.
+                return self
+                    .parent
+                    .analyze_plan(&AnyQuery::Query(request), options)
+                    .await;
+            }
+
+            let plan = self.create_plan(options).await?;
+            execute_plan(plan.clone(), Default::default())?
+                .try_collect::<Vec<_>>()
+                .await?;
+            return Ok(format_analyzed_plan(plan));
+        }
+
        let query = AnyQuery::Query(self.request.clone());
        self.parent.analyze_plan(&query, options).await
    }
@@ -1652,6 +2269,7 @@ mod tests {
        types::Float32Type,
    };
    use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
+    use datafusion_physical_plan::display::DisplayableExecutionPlan;
    use futures::{StreamExt, TryStreamExt};
    use lance_testing::datagen::{BatchGenerator, IncrementingInt32, RandomVector};
    use rand::seq::IndexedRandom;
@@ -2660,6 +3278,218 @@ mod tests {
        assert_eq!(results[0].num_columns(), 1);
    }

+    #[tokio::test]
+    async fn test_take_offsets_preserves_duplicate_multiplicity() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+
+        let results = table
+            .take_offsets(vec![5, 1, 5, 17])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .execute_with_options(QueryExecutionOptions {
+                max_batch_length: 2,
+                ..Default::default()
+            })
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+
+        assert_eq!(results.len(), 2);
+        assert!(results.iter().all(|batch| batch.num_columns() == 1));
+        let mut ids = results
+            .iter()
+            .flat_map(|batch| {
+                batch
+                    .column_by_name("id")
+                    .unwrap()
+                    .as_primitive::<Int32Type>()
+                    .values()
+                    .to_vec()
+            })
+            .collect::<Vec<_>>();
+        ids.sort_unstable();
+        assert_eq!(ids, vec![1, 5, 5, 17]);
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_plan_is_incremental() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+
+        let plan = table
+            .take_offsets(vec![5, 1, 17])
+            .create_plan(QueryExecutionOptions {
+                max_batch_length: 1,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(plan.properties().emission_type, EmissionType::Incremental);
+        let displayed = DisplayableExecutionPlan::new(plan.as_ref())
+            .indent(false)
+            .to_string();
+        assert!(displayed.contains("TakeRestoreExec"));
+        assert!(!displayed.contains("CoalescePartitionsExec"));
+    }
+
+    #[tokio::test]
+    async fn test_take_into_request_preserves_duplicate_multiplicity() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+        let request = table.take_offsets(vec![5, 5]).into_request();
+        assert_eq!(request.take_offsets, Some(vec![5, 5]));
+
+        let batches = table
+            .base_table()
+            .query(&AnyQuery::Query(request), QueryExecutionOptions::default())
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+
+        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+    }
+
+    #[test]
+    fn test_restore_take_batch_only_reorders_when_requested() {
+        let batch = RecordBatch::try_from_iter([
+            (
+                "id",
+                Arc::new(Int32Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
+            ),
+            (
+                "_rowoffset",
+                Arc::new(UInt64Array::from(vec![17, 5, 1])) as Arc<dyn Array>,
+            ),
+        ])
+        .unwrap();
+
+        let restored =
+            restore_take_batch(batch.clone(), &[5, 1, 5, 17], "_rowoffset", true, false).unwrap();
+        assert_eq!(
+            restored
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values(),
+            &[17, 5, 5, 1]
+        );
+
+        let ordered = restore_take_batch(batch, &[5, 1, 5, 17], "_rowoffset", true, true).unwrap();
+        assert_eq!(
+            ordered
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values(),
+            &[5, 1, 5, 17]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_applies_pagination_after_restoration() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+
+        let limited = table
+            .take_offsets(vec![0, 1, 0, 2])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .limit(3)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let limited = concat_batches(&limited[0].schema(), &limited).unwrap();
+        assert_eq!(limited.num_rows(), 3);
+        assert!(
+            limited
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values()
+                .iter()
+                .all(|id| [0, 1, 2].contains(id))
+        );
+
+        let offset = table
+            .take_offsets(vec![5, 1, 5, 17])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .offset(1)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let offset = concat_batches(&offset[0].schema(), &offset).unwrap();
+        assert_eq!(offset.num_rows(), 3);
+        assert!(
+            offset
+                .column_by_name("id")
+                .unwrap()
+                .as_primitive::<Int32Type>()
+                .values()
+                .iter()
+                .all(|id| [1, 5, 17].contains(id))
+        );
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_create_plan_restores_occurrences() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+        let take = table
+            .take_offsets(vec![5, 1, 5, 17])
+            .select(Select::Columns(vec!["id".to_string()]));
+
+        let plan = take
+            .create_plan(QueryExecutionOptions::default())
+            .await
+            .unwrap();
+        assert_eq!(plan.schema().fields().len(), 1);
+        assert_eq!(plan.schema().field(0).name(), "id");
+        let planned = execute_plan(plan, Default::default())
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let planned = concat_batches(&planned[0].schema(), &planned).unwrap();
+        let mut ids = planned
+            .column_by_name("id")
+            .unwrap()
+            .as_primitive::<Int32Type>()
+            .values()
+            .to_vec();
+        ids.sort_unstable();
+        assert_eq!(ids, vec![1, 5, 5, 17]);
+    }
+
+    #[tokio::test]
+    async fn test_take_offsets_plan_introspection_shows_restoration() {
+        let tmp_dir = tempdir().unwrap();
+        let table = make_test_table(&tmp_dir).await;
+        let take = table
+            .take_offsets(vec![0, 1, 0, 2])
+            .select(Select::Columns(vec!["id".to_string()]))
+            .limit(3);
+
+        let explained = take.explain_plan(false).await.unwrap();
+        assert!(explained.contains("GlobalLimitExec"));
+        assert!(explained.contains("TakeRestoreExec"));
+        assert!(!explained.contains("CoalescePartitionsExec"));
+
+        let analyzed = take.analyze_plan().await.unwrap();
+        assert!(analyzed.contains("GlobalLimitExec"));
+        assert!(analyzed.contains("TakeRestoreExec"));
+        assert!(!analyzed.contains("CoalescePartitionsExec"));
+    }
+
    #[tokio::test]
    async fn test_take_row_ids() {
        let tmp_dir = tempdir().unwrap();
@@ -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.
@@ -586,6 +593,14 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
        query: &AnyQuery,
        options: QueryExecutionOptions,
    ) -> Result<String>;
+    /// Whether [`BaseTable::analyze_plan`] is provided by a remote service.
+    ///
+    /// Client-side query wrappers use this to preserve backend metrics and
+    /// distributed-analysis options instead of replacing them with a local plan.
+    #[doc(hidden)]
+    fn analyze_plan_is_remote(&self) -> bool {
+        false
+    }

    /// Add new records to the table.
    async fn add(&self, add: AddDataBuilder) -> Result<AddResult>;
@@ -785,6 +800,12 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
    async fn drop_columns(&self, columns: &[&str]) -> Result<DropColumnsResult>;
    /// Get the version of the table.
    async fn version(&self) -> Result<u64>;
+    /// Return a new table handle pinned to the exact revision currently visible.
+    async fn checkout_current(&self) -> Result<Arc<dyn BaseTable>> {
+        Err(Error::NotSupported {
+            message: "checkout_current is not supported on this table type".into(),
+        })
+    }
    /// Checkout a specific version of the table.
    async fn checkout(&self, version: u64) -> Result<()>;
    /// Checkout a table version referenced by a tag.
@@ -1133,6 +1154,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
@@ -1618,9 +1649,9 @@ impl Table {
    /// Offsets are useful for sampling as the set of all valid offsets is easily
    /// known in advance to be [0, len(table)).
    ///
-    /// No guarantees are made regarding the order in which results are returned.  If you
-    /// desire an output order that matches the order of the given offsets, you will need
-    /// to add the row offset column to the output and align it yourself.
+    /// No guarantees are made regarding the order in which results are returned.
+    /// Repeated offsets produce repeated rows, which makes this method suitable for
+    /// sampling with replacement.
    ///
    /// Parameters
    /// ----------
@@ -1944,6 +1975,20 @@ impl Table {
        self.inner.version().await
    }

+    /// Return a new table handle pinned to the exact revision currently visible.
+    ///
+    /// This is used when asynchronous preparation must remain consistent with
+    /// the revision used for a later read.
+    #[doc(hidden)]
+    pub async fn checkout_current(&self) -> Result<Self> {
+        let inner = self.inner.checkout_current().await?;
+        Ok(Self {
+            inner,
+            database: self.database.clone(),
+            embedding_registry: self.embedding_registry.clone(),
+        })
+    }
+
    /// Checks out a specific version of the Table
    ///
    /// Any read operation on the table will now access the data at the checked out version.
@@ -3039,10 +3084,33 @@ 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)
    }

+    async fn checkout_current(&self) -> Result<Arc<dyn BaseTable>> {
+        let current = self.dataset.get().await?;
+        let dataset = dataset::DatasetConsistencyWrapper::new_time_travel(
+            current.as_ref().clone(),
+            self.read_consistency_interval,
+        );
+        Ok(Arc::new(Self {
+            dataset,
+            ..self.clone()
+        }))
+    }
+
    async fn checkout(&self, version: u64) -> Result<()> {
        self.dataset.as_time_travel(version).await
    }
@@ -4086,6 +4154,14 @@ mod tests {
                parent_list_calls: self.parent_list_calls.clone(),
            })
        }
+
+        fn wrap_paginated(
+            &self,
+            _store_prefix: &str,
+            _original: Arc<dyn object_store::list::PaginatedListStore>,
+        ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
+            None
+        }
    }

    #[tokio::test]
@@ -4189,6 +4265,14 @@ mod tests {
            self.called.store(true, Ordering::Relaxed);
            original
        }
+
+        fn wrap_paginated(
+            &self,
+            _store_prefix: &str,
+            original: Arc<dyn object_store::list::PaginatedListStore>,
+        ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
+            Some(original)
+        }
    }

    #[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();
@@ -93,7 +93,7 @@ fn requires_local_namespace_execution(query: &AnyQuery) -> bool {
    // pushing these down would silently ignore the user's setting. For use_lsm that
    // is worse than a tuning miss: MemWAL read routing lives only in `create_plan`,
    // so a pushed-down query would return stale base-only data with no error.
-    if query.base().use_lsm.is_some() {
+    if query.base().use_lsm.is_some() || query.base().take_offsets.is_some() {
        return true;
    }
    matches!(
@@ -136,6 +136,13 @@ pub async fn create_plan(
    query: &AnyQuery,
    options: QueryExecutionOptions,
 ) -> Result<Arc<dyn ExecutionPlan>> {
+    if let AnyQuery::Query(request) = query
+        && let Some(offsets) = &request.take_offsets
+    {
+        return crate::query::create_take_offsets_plan(table, request, offsets, options, false)
+            .await;
+    }
+
    let query = match query {
        AnyQuery::VectorQuery(query) => query.clone(),
        AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query.clone()),
@@ -697,6 +704,7 @@ mod tests {

    use super::*;
    use crate::query::{QueryExecutionOptions, QueryRequest};
+    use crate::table::BaseTable;

    fn fixed_size_list_array(values: Vec<f32>, dimension: i32) -> FixedSizeListArray {
        FixedSizeListArray::try_new_from_values(Float32Array::from(values), dimension).unwrap()
@@ -889,10 +897,56 @@ mod tests {

        async fn query_table(&self, _request: NsQueryTableRequest) -> lance::Result<bytes::Bytes> {
            self.query_table_calls.fetch_add(1, Ordering::SeqCst);
-            panic!("approx_mode queries must not be pushed down to namespace query_table");
+            panic!("query must not be pushed down to namespace query_table");
        }
    }

+    #[tokio::test]
+    async fn test_execute_query_pinned_snapshot_with_namespace_pushdown_runs_locally() {
+        use crate::connect;
+        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, 4, 5]))],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("test_pinned_namespace_fallback", vec![batch])
+            .execute()
+            .await
+            .unwrap();
+
+        let namespace_client = Arc::new(CountingNamespaceClient::default());
+        let mut native_table = table.as_native().unwrap().clone();
+        native_table.namespace_client = Some(namespace_client.clone());
+        native_table
+            .pushdown_operations
+            .insert(NamespaceClientPushdownOperation::QueryTable);
+
+        let snapshot = native_table.checkout_current().await.unwrap();
+        let snapshot = snapshot.as_any().downcast_ref::<NativeTable>().unwrap();
+        assert!(snapshot.dataset.time_travel_version().is_some());
+
+        let query = AnyQuery::Query(QueryRequest {
+            filter: Some(QueryFilter::Sql("id > 3".to_string())),
+            ..Default::default()
+        });
+        let stream = execute_query(snapshot, &query, QueryExecutionOptions::default())
+            .await
+            .unwrap();
+        let batches = stream.try_collect::<Vec<_>>().await.unwrap();
+
+        assert_eq!(
+            batches.iter().map(|batch| batch.num_rows()).sum::<usize>(),
+            2
+        );
+        assert_eq!(namespace_client.query_table_calls.load(Ordering::SeqCst), 0);
+    }
+
    #[tokio::test]
    async fn test_execute_query_approx_mode_with_namespace_pushdown_runs_locally() {
        use crate::connect;
@@ -1009,6 +1063,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};
@@ -298,7 +298,7 @@ async fn build_read_context(
    for shard_id in shard_ids {
        let manifest_store =
            ShardManifestStore::new(store.clone(), &base_path, shard_id, scan_batch_size);
-        if let Some(manifest) = manifest_store.read_latest().await? {
+        if let Some(manifest) = manifest_store.latest().await? {
            snapshots.push(snapshot_from_manifest(shard_id, &manifest, &exclude));
        }
    }
Author	SHA1	Message	Date
Gatefixer	029930b412	Merge remote-tracking branch 'refs/remotes/origin/main' into gatekeeper/fix-2820-1 # Conflicts: # rust/lancedb/src/remote/table.rs	2026-08-26 05:43:48 +00: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
LanceDB Robot	8083232dd5	chore: update lance dependency to v12.0.0-beta.1 (#4055 ) Updates the Lance Rust workspace dependencies and Java lance-core dependency to [v12.0.0-beta.1](https://github.com/lance-format/lance/releases/tag/v12.0.0-beta.1). Includes compatibility updates for the renamed shard-manifest API and paginated object-store wrappers.	2026-08-25 16:49:17 -07:00
lancedb-gatefixer[bot]	302b21aa94	test(node): cover nested PDF metadata queries (#3827 ) ## Summary - add an end-to-end Node regression matching LangChain PDFLoader metadata - verify create/query round trips rich nested `loc` and `pdf.info` fields against the currently configured Apache Arrow peer ## Root cause LanceDB v0.14 delegated nested object inference to Apache Arrow. Nested strings were dictionary-encoded with colliding dictionary IDs, so serializing query results as an IPC file failed with a dictionary-replacement error. Current `main` recursively infers nested fields and avoids those invalid dictionaries, but the reported LangChain path had no end-to-end regression coverage. ## Validation - `pnpm build` - `pnpm lint` - `pnpm run docs` - `pnpm test --runInBand` (678 passed, 5 skipped) Fixes #1963 <!-- lance-gatekeeper-fix:v1 agent=bf8d489db7db2e17678b143f9f0a36d2 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-26 06:42:29 +08:00
lancedb-gatefixer[bot]	35b5d015ac	fix(node): preserve embedding registration in server bundles (#3806 ) ## Summary - lazily initialize built-in OpenAI and Hugging Face providers when consumers call the public embedding registry API - choose automatic vector versus FTS search from embedding metadata on a fresh pinned table revision for every execution - expose automatic string searches as an `AutoQuery` with only operations common to both native query families - keep the registry shared and built-in registration safe across duplicated module graphs ## Root cause Nitro treats dependency modules as side-effect-free and removes the bare OpenAI provider import from its generated route. Registration therefore never runs, so `getRegistry().get("openai")` remains undefined even when the registry itself is shared globally. Bundlers may also duplicate the provider and registry module graphs. The public embedding entry point now initializes built-in providers only when `getRegistry()` is explicitly called, keeping initialization on a live path that Nitro retains. Each terminal automatic-search execution pins the exact table revision visible at dispatch, reads embedding metadata and computes an embedding from that snapshot, replays the builder operations, and constructs and executes the selected native query against the same snapshot. Pinned native snapshots execute locally when namespace pushdown cannot carry their revision, while remote snapshots are seeded directly from one version-and-schema response. The public `AutoQuery` builder exposes only the operations shared by FTS and vector search, so runtime class narrowing cannot expose invalid vector-only methods. Repeated built-in registration replaces stale constructors from duplicated module graphs while public `register()` retains its duplicate-alias error. ## Validation - `cargo fmt --all` - `cargo check --quiet --features remote --tests --examples` - `cargo clippy --quiet --features remote --tests --examples` - `pnpm build` - `pnpm lint` - `pnpm run docs` - `pnpm test --runInBand` (783 passed, 5 skipped) - serial examples suite with a local OpenAI mock (11 passed), including `sentence-transformers.test.ts` - packaged Nitro 2.13.4 server route using the reported imports returned `{"registered":true}` - fresh-process FTS fixture initialized both public built-ins and confirmed automatic string search still returned the indexed row - schema-consistency regressions cover read-consistency refresh, checkout, checkoutLatest, restore, runtime class narrowing, concurrent overwrite during embedding computation, and reused automatic-search builders - focused regressions confirm pinned native snapshots bypass unversioned namespace pushdown and remote snapshots use one describe request Fixes #2429 <!-- lance-gatekeeper-fix:v1 agent=2adf0f21b8bfb634606ed8897a849e30 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-26 04:17:32 +08:00
lancedb-gatefixer[bot]	a57fb68891	docs(python): fix Azure storage options examples (#3899 ) ## Summary - document that Azure Blob Storage credentials can be passed directly through `storage_options` - provide valid quoted `account_name` and `account_key` examples for both sync and async Python connections - execute the option dictionaries during doctests so the original unquoted-key mistake is caught ## Root cause The historical Python storage guide used `account_name` and `account_key` as bare identifiers in dictionary literals. Following that example either raised `NameError` or, when those names were predefined, produced incorrect option keys. The runtime already accepts direct Azure credentials, but the current Python API reference did not contain a corrected Azure example. ## Validation - `python/.venv/bin/ruff format --check python/python/lancedb/__init__.py` - `python/.venv/bin/ruff check .` - `cd python && uv run --no-sync pytest --doctest-modules python/lancedb/__init__.py -q` - `cd python && uv run --no-sync pytest python/tests/test_import.py -q` Fixes #2236 <!-- lance-gatekeeper-fix:v1 agent=7a7a9e009eb2ebe52ac9b1adf2e8afb2 generation=1 --> Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-26 00:31:14 +08:00
Gatefixer	f936f65626	Merge remote-tracking branch 'refs/remotes/origin/main' into gatekeeper/fix-2820-1	2026-08-24 20:19:27 +00:00
Gatefixer	4bbb368eee	fix: preserve streaming take conversions	2026-08-24 19:37:11 +00:00
Gatefixer	1ea02116e0	style(python): format duplicate offset test	2026-08-24 18:04:29 +00:00
Gatefixer	241d0604d3	Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2820-1	2026-08-24 17:41:48 +00:00
Gatefixer	d87d69c181	fix: retain unordered take semantics	2026-08-24 17:25:13 +00:00
Gatefixer	0eda8ac32c	fix: delegate remote take analysis to server	2026-08-24 16:13:07 +00:00
Gatefixer	23d243346e	Merge origin/main into gatekeeper/fix-2820-1	2026-08-22 01:06:03 +00:00
Gatefixer	67d6799f4a	fix: preserve remote take analysis metrics	2026-08-22 01:04:21 +00:00
Gatefixer	17e13e0e6d	fix: align take plan introspection	2026-08-22 00:33:36 +00:00
Gatefixer	7305b5a120	fix: restore take occurrences in query plans	2026-08-21 23:35:07 +00:00
Gatefixer	578253e892	fix: preserve duplicate take offsets	2026-08-21 22:50:46 +00:00