feat: expose list-element FTS document granularity (#4050 )

## Summary LanceDB could not request Lance's list-element FTS document granularity through Python or Remote APIs, and generic nested-field resolution exposed Arrow's internal `item` segment instead of the public field path. This exposes typed `row | list_element` configuration for Python FTS index creation and match/phrase queries, preserves `_doc_index`, and keeps nested FTS paths public (for example, `docs.content`). Remote list-element requests require server API version 0.6.0 so older servers cannot silently execute them with row semantics; explicit row requests remain compatible. ## Compatibility Omitted index and query parameters retain row granularity. Remote row index creation omits the new wire field. ## Tracking [ENT-2342](https://linear.app/lancedb/issue/ENT-2342/expose-list-element-fts-document-granularity-end-to-end)
feat: declare conda environments on Functions (#4057 )
2026-08-26 16:08:43 +00:00 · 2026-08-26 23:27:39 +08:00 · 2026-08-26 06:29:43 -07:00 · 2026-08-26 14:56:57 +08:00 · 2026-08-26 12:54:36 +08:00 · 2026-08-25 21:56:26 -05:00
50 changed files with 4087 additions and 619 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)

 ***

@@ -1292,18 +1292,6 @@ abstract updateFieldMetadata(updates): Promise<UpdateFieldMetadataResult>

 Update per-field (column) metadata.

-The following keys are treated specially, by convention, and should be
-used when appropriate:
-
- `lancedb:description`: for a human-readable description of a field.
- `lancedb:tag:<name>`: for a user-defined key-value tag, where the suffix
-  names the tag category; e.g. `lancedb:tag:model: "clip"`.
- `lancedb:logical-column`: for a column grouping; e.g. `feature_v1` and
-  `feature_v2` might be in the same logical column.
- `lancedb:status`: for status options (`production`, `candidate`,
-  `deprecated`, `archived`) to designate the current life cycle state of
-  this column.
-
 #### Parameters

 * **updates**: [`FieldMetadataUpdate`](../interfaces/FieldMetadataUpdate.md)[]
@@ -18,6 +18,7 @@

 ## Classes

+- [AutoQuery](classes/AutoQuery.md)
 - [BooleanQuery](classes/BooleanQuery.md)
 - [BoostQuery](classes/BoostQuery.md)
 - [BranchContents](classes/BranchContents.md)
@@ -17,8 +17,7 @@ metadata: Record<string, null | string>;
 ```

 Metadata key/value pairs. Merged into the field's existing metadata by
-default; a value of `null` deletes that key. See
-[Table.updateFieldMetadata](../classes/Table.md#updatefieldmetadata) for the conventional `lancedb:*` keys.
+default; a value of `null` deletes that key.

 ***

@@ -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();
@@ -159,6 +159,8 @@ and combined with [BooleanQuery][lancedb.query.BooleanQuery].

 ::: lancedb.query.FullTextOperator

+::: lancedb.query.DocumentGranularity
+
 ::: lancedb.query.Occur

 ## Embeddings
@@ -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.
   *
@@ -628,18 +630,6 @@ export abstract class Table {

  /**
   * Update per-field (column) metadata.
-   *
-   * The following keys are treated specially, by convention, and should be
-   * used when appropriate:
-   *
-   * - `lancedb:description`: for a human-readable description of a field.
-   * - `lancedb:tag:<name>`: for a user-defined key-value tag, where the suffix
-   *   names the tag category; e.g. `lancedb:tag:model: "clip"`.
-   * - `lancedb:logical-column`: for a column grouping; e.g. `feature_v1` and
-   *   `feature_v2` might be in the same logical column.
-   * - `lancedb:status`: for status options (`production`, `candidate`,
-   *   `deprecated`, `archived`) to designate the current life cycle state of
-   *   this column.
   * @param {FieldMetadataUpdate[]} updates One or more per-field updates. Each
   * update's metadata is merged into the field's existing metadata by default;
   * a value of `null` deletes that key, and `replace: true` swaps the whole map.
@@ -987,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);
  }
@@ -1172,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");
@@ -1187,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);
      });
    }

@@ -1550,8 +1555,7 @@ export interface FieldMetadataUpdate {
  path: string;
  /**
   * Metadata key/value pairs. Merged into the field's existing metadata by
-   * default; a value of `null` deletes that key. See
-   * {@link Table.updateFieldMetadata} for the conventional `lancedb:*` keys.
+   * default; a value of `null` deletes that key.
   */
  metadata: Record<string, string | null>;
  /** If true, replace the field's entire metadata map instead of merging. */
@@ -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
@@ -222,6 +222,7 @@ class PythonEnvironmentSpec(_RemoteValue):

    kind: str
    packages: tuple[str, ...] = ()
+    channels: tuple[str, ...] = ()
    path: Optional[str] = None
    modules: tuple[str, ...] = ()
    image: Optional[str] = None
@@ -909,13 +910,25 @@ class UdfDefinition:
        pip: tuple[str, ...],
        env: Mapping[str, str],
        python_version: Optional[str],
+        conda: tuple[str, ...] = (),
+        conda_channels: tuple[str, ...] = (),
    ):
        function_name = name or function.__name__
        if not _FUNCTION_NAME.fullmatch(function_name):
            raise ValueError(f"invalid Function name: {function_name!r}")
-        packages = tuple(sorted(set(pip)))
+        if pip and conda:
+            raise ValueError("a Function environment is pip or conda, not both")
+        if conda_channels and not conda:
+            raise ValueError("conda_channels requires conda packages")
+        packages = tuple(sorted(set(conda if conda else pip)))
        if any(not package or package != package.strip() for package in packages):
-            raise ValueError("pip requirements must be non-empty and trimmed")
+            raise ValueError("package requirements must be non-empty and trimmed")
+        if conda:
+            environment_spec = PythonEnvironmentSpec(
+                kind="conda", packages=packages, channels=tuple(conda_channels)
+            )
+        else:
+            environment_spec = PythonEnvironmentSpec(kind="pip", packages=packages)
        environment = dict(env)
        if any(
            not isinstance(key, str) or not isinstance(value, str)
@@ -929,7 +942,7 @@ class UdfDefinition:
            kind="python",
            python_version=python_version
            or f"{sys.version_info.major}.{sys.version_info.minor}",
-            environment=PythonEnvironmentSpec(kind="pip", packages=packages),
+            environment=environment_spec,
            env=environment,
        )
        self._function = function
@@ -976,6 +989,8 @@ def udf(
    pip: tuple[str, ...] | list[str] = (),
    env: Optional[Mapping[str, str]] = None,
    python_version: Optional[str] = None,
+    conda: tuple[str, ...] | list[str] = (),
+    conda_channels: tuple[str, ...] | list[str] = (),
 ) -> Callable[[Callable[..., Any]], UdfDefinition]: ...


@@ -988,6 +1003,8 @@ def udf(
    pip: tuple[str, ...] | list[str] = (),
    env: Optional[Mapping[str, str]] = None,
    python_version: Optional[str] = None,
+    conda: tuple[str, ...] | list[str] = (),
+    conda_channels: tuple[str, ...] | list[str] = (),
 ):
    """Prepare a scalar Python callable for remote Function registration.

@@ -1010,6 +1027,10 @@ def udf(
        provided together with ``input_schema``.
    pip : sequence of str, optional
        Pip requirements for the remote environment.
+    conda : sequence of str, optional
+        Conda packages for the remote environment, instead of ``pip``.
+    conda_channels : sequence of str, optional
+        Conda channels in priority order; requires ``conda``.
    env : mapping of str to str, optional
        Environment variables included in the Function definition.
    python_version : str, optional
@@ -1049,6 +1070,8 @@ def udf(
            pip=tuple(pip),
            env={} if env is None else env,
            python_version=python_version,
+            conda=tuple(conda),
+            conda_channels=tuple(conda_channels),
        )

    if function is None:
@@ -7,6 +7,7 @@ from typing import List, Literal, Optional
 from ._lancedb import (
    IndexConfig,
 )
+from .query import DocumentGranularity
 from .types import BaseTokenizerType

 lang_mapping = {
@@ -121,6 +122,11 @@ class FTS:

    >>> config = FTS(block_size=256)

+    Create an index that treats each deepest-list element as one document:
+
+    >>> from lancedb.query import DocumentGranularity
+    >>> config = FTS(document_granularity=DocumentGranularity.LIST_ELEMENT)
+
    Attributes
    ----------
    with_position : bool, default False
@@ -172,6 +178,11 @@ class FTS:
        roughly half of the available CPU cores. The effective value is
        limited by the available compute capacity. This build-only setting is
        not persisted with the index and does not apply to remote tables.
+    document_granularity : DocumentGranularity, default ROW
+        ``ROW`` treats the selected text in one table row as one document.
+        ``LIST_ELEMENT`` treats each element of the deepest list on the indexed
+        field path as one document and returns its physical coordinates in
+        ``_doc_index`` for matching queries.

    Notes
    -----
@@ -196,6 +207,7 @@ class FTS:
    custom_stop_words: Optional[List[str]] = None
    memory_limit: Optional[int] = None
    num_workers: Optional[int] = None
+    document_granularity: DocumentGranularity = DocumentGranularity.ROW


@dataclass
@@ -375,6 +375,13 @@ class FullTextOperator(str, Enum):
    OR = "OR"


+class DocumentGranularity(str, Enum):
+    """The unit treated as one full-text-search document."""
+
+    ROW = "row"
+    LIST_ELEMENT = "list_element"
+
+
 class Occur(str, Enum):
    SHOULD = "SHOULD"
    MUST = "MUST"
@@ -478,6 +485,10 @@ class MatchQuery(FullTextQuery):
    prefix_length : int, optional
        The number of beginning characters being unchanged for fuzzy matching.
        This is useful to achieve prefix matching.
+    document_granularity : DocumentGranularity, optional
+        Explicitly select row or deepest-list-element documents. If omitted,
+        the indexed granularity is inferred. When both granularities are indexed
+        for the field, this must be specified. With no index, row granularity is used.
    """

    query: str
@@ -487,6 +498,9 @@ class MatchQuery(FullTextQuery):
    max_expansions: int = pydantic.Field(50, kw_only=True)
    operator: FullTextOperator = pydantic.Field(FullTextOperator.OR, kw_only=True)
    prefix_length: int = pydantic.Field(0, kw_only=True)
+    document_granularity: Optional[DocumentGranularity] = pydantic.Field(
+        None, kw_only=True
+    )

    def query_type(self) -> FullTextQueryType:
        return FullTextQueryType.MATCH
@@ -503,11 +517,20 @@ class PhraseQuery(FullTextQuery):
        The query string to match against.
    column : str
        The name of the column to match against.
+    slop : int, default 0
+        The maximum number of intervening positions permitted in the phrase.
+    document_granularity : DocumentGranularity, optional
+        Explicitly select row or deepest-list-element documents. If omitted,
+        the indexed granularity is inferred. When both granularities are indexed
+        for the field, this must be specified. With no index, row granularity is used.
    """

    query: str
    column: str
    slop: int = pydantic.Field(0, kw_only=True)
+    document_granularity: Optional[DocumentGranularity] = pydantic.Field(
+        None, kw_only=True
+    )

    def query_type(self) -> FullTextQueryType:
        return FullTextQueryType.MATCH_PHRASE
@@ -61,6 +61,7 @@ from lancedb.table import _normalize_progress

 from ..query import (
    AnalyzePlanDistributedMetrics,
+    DocumentGranularity,
    LanceQueryBuilder,
    LanceTakeQueryBuilder,
    LanceVectorQueryBuilder,
@@ -349,6 +350,7 @@ class RemoteTable(Table):
        ngram_max_length: int = 3,
        prefix_only: bool = False,
        block_size: int = 128,
+        document_granularity: DocumentGranularity = DocumentGranularity.ROW,
        name: Optional[str] = None,
    ):
        """Create a full-text search index on a column.
@@ -371,6 +373,7 @@ class RemoteTable(Table):
            ngram_max_length=ngram_max_length,
            prefix_only=prefix_only,
            block_size=block_size,
+            document_granularity=document_granularity,
        )
        LOOP.run(
            self._table.create_index(
@@ -85,6 +85,7 @@ from .query import (
    AsyncQuery,
    AsyncTakeQuery,
    AsyncVectorQuery,
+    DocumentGranularity,
    FullTextQuery,
    LanceEmptyQueryBuilder,
    LanceFtsQueryBuilder,
@@ -1168,6 +1169,7 @@ class Table(ABC):
        ngram_max_length: int = 3,
        prefix_only: bool = False,
        block_size: int = 128,
+        document_granularity: DocumentGranularity = DocumentGranularity.ROW,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
    ):
@@ -1246,6 +1248,11 @@ class Table(ABC):
            The number of documents per compressed posting block. Must be 128
            or 256. A value of 256 uses the experimental FTS V3 format and
            may introduce breaking changes.
+        document_granularity: DocumentGranularity, default ROW
+            ``ROW`` treats the selected text in one table row as one document.
+            ``LIST_ELEMENT`` treats each element of the deepest list on the field
+            path as one document and returns its physical coordinates in
+            ``_doc_index`` for matching queries.
        wait_timeout: timedelta, optional
            The timeout to wait if indexing is asynchronous.
        name: str, optional
@@ -2120,25 +2127,12 @@ class Table(ABC):
        ----------
        updates : dict
            One or more dicts, each with:
-
            - "path": str — dot-path to the field (e.g. "embedding" or "a.b.c").
            - "metadata": dict[str, str | None] — keys to set; a value of ``None``
              deletes that key.
            - "replace": bool, optional — replace the field's whole metadata map
              instead of merging (default False).

-            The following keys are treated specially, by convention, and should
-            be used when appropriate:
-
-            - "lancedb:description": for a human-readable description of a field.
-            - ``"lancedb:tag:<name>"`` for a user-defined key-value tag, where the
-                suffix names the tag category; e.g. "lancedb:tag:model": "clip".
-            - "lancedb:logical-column" for a column grouping; e.g. "feature_v1"
-                and "feature_v2" might be in the same logical column.
-            - "lancedb:status" for status options ("production", "candidate",
-                "deprecated", "archived") to designate the current life cycle
-                state of this column.
-
        Returns
        -------
        UpdateFieldMetadataResult
@@ -3286,6 +3280,7 @@ class LanceTable(Table):
        ngram_max_length: int = 3,
        prefix_only: bool = False,
        block_size: int = 128,
+        document_granularity: DocumentGranularity = DocumentGranularity.ROW,
        name: Optional[str] = None,
    ):
        """Create a full-text search index on a column.
@@ -3337,7 +3332,11 @@ class LanceTable(Table):
            tokenizer_configs = self.infer_tokenizer_configs(tokenizer_name)
            tokenizer_configs["custom_stop_words"] = custom_stop_words

-        config = FTS(block_size=block_size, **tokenizer_configs)
+        config = FTS(
+            block_size=block_size,
+            document_granularity=document_granularity,
+            **tokenizer_configs,
+        )

        try:
            LOOP.run(
@@ -69,6 +69,26 @@ def _run_packaged(definition, *args):
    return namespace[definition.registration_request.artifact.entrypoint](*args)


+def test_udf_conda_environment():
+    @udf(conda=["scipy", "numpy"], conda_channels=["conda-forge", "defaults"])
+    def halve(value: float) -> float:
+        return value / 2
+
+    request = json.loads(halve.registration_request.to_canonical_json())
+    assert request["runtime"]["environment"] == {
+        "kind": "conda",
+        "packages": ["numpy", "scipy"],
+        "channels": ["conda-forge", "defaults"],
+    }
+    pip_request = json.loads(normalize_score.registration_request.to_canonical_json())
+    assert "channels" not in pip_request["runtime"]["environment"]
+
+    with pytest.raises(ValueError, match="not both"):
+        udf(name="both", pip=["numpy"], conda=["numpy"])(lambda value: value)
+    with pytest.raises(ValueError, match="requires conda"):
+        udf(name="channels", conda_channels=["conda-forge"])(lambda value: value)
+
+
 def test_udf_packages_attribute_access_and_body_imports():
    @udf
    def word_norm(body: str) -> float:
@@ -25,6 +25,7 @@ from lancedb.db import DBConnection
 from lancedb.index import FTS
 from lancedb.query import (
    BoostQuery,
+    DocumentGranularity,
    MatchQuery,
    MultiMatchQuery,
    PhraseQuery,
@@ -245,6 +246,55 @@ def test_create_inverted_index_rejects_invalid_block_size(table):
        table.create_index("text", config=FTS(block_size=129))


+def test_list_element_document_granularity(tmp_path):
+    docs_type = pa.list_(pa.struct([pa.field("content", pa.string())]))
+    docs = pa.array(
+        [
+            [
+                {"content": "alpha beta"},
+                None,
+                {"content": ""},
+                {"content": "the and"},
+                {"content": "alpha beta"},
+            ]
+        ],
+        type=docs_type,
+    )
+    table = ldb.connect(tmp_path).create_table(
+        "list_element_docs", pa.table({"id": [0], "docs": docs})
+    )
+    row_table = ldb.connect(tmp_path).create_table(
+        "row_docs", pa.table({"id": [0], "docs": docs})
+    )
+    row_table.create_index("docs.content", config=FTS())
+    row_result = row_table.search(MatchQuery("alpha", "docs.content")).to_arrow()
+    assert row_result.num_rows == 1
+    assert "_doc_index" not in row_result.column_names
+
+    granularity = DocumentGranularity.LIST_ELEMENT
+    table.create_index(
+        "docs.content",
+        config=FTS(with_position=True, document_granularity=granularity),
+    )
+    assert table.list_indices()[0].columns == ["docs.content"]
+
+    def coordinates(query):
+        result = table.search(query).limit(10).to_arrow()
+        doc_index_type = result.schema.field("_doc_index").type
+        assert pa.types.is_list(doc_index_type)
+        assert doc_index_type.value_type == pa.uint32()
+        return sorted(result["_doc_index"].to_pylist())
+
+    assert coordinates(
+        MatchQuery("alpha", "docs.content", document_granularity=granularity)
+    ) == [[0], [4]]
+    assert coordinates(
+        PhraseQuery("alpha beta", "docs.content", document_granularity=granularity)
+    ) == [[0], [4]]
+    assert coordinates(MatchQuery("alpha", "docs.content")) == [[0], [4]]
+    assert FTS().document_granularity is DocumentGranularity.ROW
+
+
 def test_create_inverted_index_respects_build_memory_limit(table):
    with pytest.raises(ValueError, match="exceeds worker memory limit"):
        table.create_index(
@@ -1089,6 +1139,20 @@ def test_fts_query_to_json():
    )
    assert json_str == expected

+    # Test MatchQuery with list-element document granularity
+    match_query = MatchQuery(
+        "hello world",
+        "text",
+        document_granularity=DocumentGranularity.LIST_ELEMENT,
+    )
+    json_str = match_query.to_json()
+    expected = (
+        '{"match":{"column":"text","terms":"hello world","boost":1.0,'
+        '"fuzziness":0,"max_expansions":50,"operator":"Or","prefix_length":0,'
+        '"document_granularity":"list_element"}}'
+    )
+    assert json_str == expected
+
    # Test MatchQuery with options
    match_query = MatchQuery("puppy", "text", fuzziness=2, boost=1.5, prefix_length=3)
    json_str = match_query.to_json()
@@ -1098,6 +1162,19 @@ def test_fts_query_to_json():
    )
    assert json_str == expected

+    # Test PhraseQuery with list-element document granularity
+    phrase_query = PhraseQuery(
+        "quick brown fox",
+        "title",
+        document_granularity=DocumentGranularity.LIST_ELEMENT,
+    )
+    json_str = phrase_query.to_json()
+    expected = (
+        '{"phrase":{"column":"title","terms":"quick brown fox","slop":0,'
+        '"document_granularity":"list_element"}}'
+    )
+    assert json_str == expected
+
    # Test PhraseQuery
    phrase_query = PhraseQuery("quick brown fox", "title")
    json_str = phrase_query.to_json()
@@ -1618,6 +1618,49 @@ def test_query_sync_fts():
        )


+def test_query_sync_fts_document_granularity():
+    from lancedb.query import DocumentGranularity, MatchQuery
+
+    def handler(body):
+        assert body == {
+            "full_text_query": {
+                "query": {
+                    "match": {
+                        "column": "docs.content",
+                        "terms": "alpha",
+                        "boost": 1.0,
+                        "fuzziness": 0,
+                        "max_expansions": 50,
+                        "operator": "Or",
+                        "prefix_length": 0,
+                        "document_granularity": "list_element",
+                    }
+                }
+            },
+            "k": 10,
+            "prefilter": True,
+            "vector": [],
+            "version": None,
+        }
+        return pa.table(
+            {
+                "id": [1, 1],
+                "_doc_index": pa.array([[0], [4]], type=pa.list_(pa.uint32())),
+            }
+        )
+
+    with query_test_table(handler, server_version=Version("0.6.0")) as table:
+        result = table.search(
+            MatchQuery(
+                "alpha",
+                "docs.content",
+                document_granularity=DocumentGranularity.LIST_ELEMENT,
+            )
+        ).to_arrow()
+
+    assert result["_doc_index"].to_pylist() == [[0], [4]]
+
+
 def test_query_sync_hybrid():
    def handler(body):
        if "full_text_query" in body:
@@ -4,6 +4,7 @@

 import asyncio
 import copy
+from concurrent.futures import ThreadPoolExecutor
 from datetime import timedelta
 import threading

@@ -86,6 +87,25 @@ def test_s3_lifecycle(s3_bucket: str):
    asyncio.run(test())


+@pytest.mark.s3_test
+def test_concurrent_open_table(s3_bucket: str):
+    uri = f"s3://{s3_bucket}/test_concurrent_open_table"
+    db = lancedb.connect(uri, storage_options=copy.copy(CONFIG))
+    db.create_table("test", pa.table({"x": [1, 2, 3]}))
+
+    num_workers = 32
+    barrier = threading.Barrier(num_workers)
+
+    def open_and_count(_):
+        barrier.wait()
+        return db.open_table("test").count_rows()
+
+    with ThreadPoolExecutor(max_workers=num_workers) as pool:
+        row_counts = list(pool.map(open_and_count, range(num_workers)))
+
+    assert row_counts == [3] * num_workers
+
+
@pytest.fixture()
 def kms_key():
    kms = get_boto3_client("kms", endpoint_url=CONFIG["aws_endpoint"])
@@ -8,7 +8,7 @@ use lancedb::index::vector::{
 };
 use lancedb::index::{
    Index as LanceDbIndex,
-    scalar::{BTreeIndexBuilder, FmIndexBuilder, FtsIndexBuilder},
+    scalar::{BTreeIndexBuilder, DocumentGranularity, FmIndexBuilder, FtsIndexBuilder},
 };
 use pyo3::IntoPyObject;
 use pyo3::types::PyStringMethods;
@@ -60,7 +60,11 @@ pub fn extract_index_params(source: &Option<Bound<'_, PyAny>>) -> PyResult<Lance
                    .ngram_min_length(params.ngram_min_length)
                    .ngram_max_length(params.ngram_max_length)
                    .ngram_prefix_only(params.prefix_only)
-                    .custom_stop_words(params.custom_stop_words);
+                    .custom_stop_words(params.custom_stop_words)
+                    .document_granularity(
+                        DocumentGranularity::try_from(params.document_granularity.as_str())
+                            .map_err(|err| PyValueError::new_err(err.to_string()))?,
+                    );
                if let Some(memory_limit) = params.memory_limit {
                    inner_opts = inner_opts.memory_limit_mb(memory_limit);
                }
@@ -221,6 +225,7 @@ struct FtsParams {
    block_size: usize,
    memory_limit: Option<u64>,
    num_workers: Option<usize>,
+    document_granularity: String,
 }

 #[derive(FromPyObject)]
@@ -481,6 +486,7 @@ mod tests {
    block_size = 128
    memory_limit = 2048
    num_workers = 7
+    document_granularity = 'row'

 config = FTS()",
                None,
@@ -16,8 +16,8 @@ use arrow::pyarrow::FromPyArrow;
 use arrow::pyarrow::IntoPyArrow;
 use arrow::pyarrow::ToPyArrow;
 use lancedb::index::scalar::{
-    BooleanQuery, BoostQuery, FtsQuery, FullTextSearchQuery, MatchQuery, MultiMatchQuery, Occur,
-    Operator, PhraseQuery,
+    BooleanQuery, BoostQuery, DocumentGranularity, FtsQuery, FullTextSearchQuery, MatchQuery,
+    MultiMatchQuery, Occur, Operator, PhraseQuery,
 };
 use lancedb::query::AnalyzePlanDistributedMetrics;
 use lancedb::query::QueryBase;
@@ -76,8 +76,16 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
                let max_expansions = ob.getattr("max_expansions")?.extract()?;
                let operator = ob.getattr("operator")?.extract::<String>()?;
                let prefix_length = ob.getattr("prefix_length")?.extract()?;
+                let document_granularity = ob
+                    .getattr("document_granularity")?
+                    .extract::<Option<String>>()?
+                    .map(|value| {
+                        DocumentGranularity::try_from(value.as_str())
+                            .map_err(|err| PyValueError::new_err(err.to_string()))
+                    })
+                    .transpose()?;

-                Ok(Self(
+                let mut query =
                    MatchQuery::new(query)
                        .with_column(Some(column))
                        .with_boost(boost)
@@ -86,21 +94,32 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
                        .with_operator(Operator::try_from(operator.as_str()).map_err(|e| {
                            PyValueError::new_err(format!("Invalid operator: {}", e))
                        })?)
-                        .with_prefix_length(prefix_length)
-                        .into(),
-                ))
+                        .with_prefix_length(prefix_length);
+                if let Some(document_granularity) = document_granularity {
+                    query = query.with_document_granularity(document_granularity);
+                }
+                Ok(Self(query.into()))
            }
            "PhraseQuery" => {
                let query = ob.getattr("query")?.extract()?;
                let column = ob.getattr("column")?.extract()?;
                let slop = ob.getattr("slop")?.extract()?;
+                let document_granularity = ob
+                    .getattr("document_granularity")?
+                    .extract::<Option<String>>()?
+                    .map(|value| {
+                        DocumentGranularity::try_from(value.as_str())
+                            .map_err(|err| PyValueError::new_err(err.to_string()))
+                    })
+                    .transpose()?;

-                Ok(Self(
-                    PhraseQuery::new(query)
-                        .with_column(Some(column))
-                        .with_slop(slop)
-                        .into(),
-                ))
+                let mut query = PhraseQuery::new(query)
+                    .with_column(Some(column))
+                    .with_slop(slop);
+                if let Some(document_granularity) = document_granularity {
+                    query = query.with_document_granularity(document_granularity);
+                }
+                Ok(Self(query.into()))
            }
            "BoostQuery" => {
                let positive: Self = ob.getattr("positive")?.extract()?;
@@ -167,6 +186,13 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
                kwargs.set_item("max_expansions", query.max_expansions)?;
                kwargs.set_item::<_, &str>("operator", query.operator.into())?;
                kwargs.set_item("prefix_length", query.prefix_length)?;
+                if let Some(document_granularity) = query.document_granularity {
+                    let value = match document_granularity {
+                        DocumentGranularity::Row => "row",
+                        DocumentGranularity::ListElement => "list_element",
+                    };
+                    kwargs.set_item("document_granularity", value)?;
+                }
                namespace
                    .getattr(intern!(py, "MatchQuery"))?
                    .call((query.terms, query.column.unwrap()), Some(&kwargs))
@@ -174,6 +200,13 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
            FtsQuery::Phrase(query) => {
                let kwargs = PyDict::new(py);
                kwargs.set_item("slop", query.slop)?;
+                if let Some(document_granularity) = query.document_granularity {
+                    let value = match document_granularity {
+                        DocumentGranularity::Row => "row",
+                        DocumentGranularity::ListElement => "list_element",
+                    };
+                    kwargs.set_item("document_granularity", value)?;
+                }
                namespace
                    .getattr(intern!(py, "PhraseQuery"))?
                    .call((query.terms, query.column.unwrap()), Some(&kwargs))
@@ -1476,7 +1476,7 @@ mod tests {
    use crate::table::{AnyQuery, WriteOptions};
    use arrow_array::{Int32Array, RecordBatch, StringArray};
    use arrow_schema::{DataType, Field, Schema, SchemaRef};
-    use futures::{TryStreamExt, stream::once};
+    use futures::{TryStreamExt, future::try_join_all, stream::once};
    use std::path::PathBuf;
    use std::sync::Arc;
    use std::time::Duration;
@@ -1614,6 +1614,59 @@ mod tests {
        );
    }

+    #[tokio::test]
+    async fn test_concurrent_open_table_reuses_connection_object_store() {
+        let tempdir = tempdir().unwrap();
+        let uri = tempdir.path().to_str().unwrap();
+        let session = Arc::new(lance::session::Session::default());
+        let request = ConnectRequest {
+            uri: uri.to_string(),
+            #[cfg(feature = "remote")]
+            client_config: Default::default(),
+            options: Default::default(),
+            namespace_client_properties: Default::default(),
+            manifest_enabled: false,
+            read_consistency_interval: None,
+            session: Some(session.clone()),
+        };
+        let db = ListingDatabase::connect_with_options(&request)
+            .await
+            .unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        db.create_table(CreateTableRequest {
+            name: "test".to_string(),
+            namespace_path: vec![],
+            data: Box::new(RecordBatch::new_empty(schema)) as Box<dyn Scannable>,
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        })
+        .await
+        .unwrap();
+
+        let before = session.store_registry().stats();
+        let opened_tables = try_join_all((0..32).map(|_| {
+            db.open_table(OpenTableRequest {
+                name: "test".to_string(),
+                namespace_path: vec![],
+                index_cache_size: None,
+                lance_read_params: None,
+                location: None,
+                namespace_client: None,
+                managed_versioning: None,
+            })
+        }))
+        .await
+        .unwrap();
+        let after = session.store_registry().stats();
+
+        assert_eq!(opened_tables.len(), 32);
+        assert_eq!(after.misses, before.misses);
+        assert_eq!(after.active_stores, before.active_stores);
+        assert!(after.hits >= before.hits + 32);
+    }
+
    #[tokio::test]
    async fn test_listing_database_root_ops_do_not_create_manifest() {
        let tempdir = tempdir().unwrap();
@@ -186,6 +186,9 @@ pub struct PythonEnvironmentSpec {
    pub kind: String,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub packages: Vec<String>,
+    /// Conda channels in priority order; conda environments only.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub channels: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub path: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
@@ -583,3 +586,26 @@ impl RefreshColumnResult {
 }

 impl_json!(RefreshColumnResult);
+
+#[cfg(test)]
+mod conda_environment_tests {
+    use super::PythonEnvironmentSpec;
+
+    #[test]
+    fn conda_channels_round_trip_and_pip_stays_bare() {
+        let conda: PythonEnvironmentSpec = serde_json::from_str(
+            r#"{"kind":"conda","packages":["numpy"],"channels":["conda-forge"]}"#,
+        )
+        .unwrap();
+        assert_eq!(conda.channels, ["conda-forge"]);
+        assert!(
+            serde_json::to_string(&conda)
+                .unwrap()
+                .contains(r#""channels":["conda-forge"]"#)
+        );
+
+        let pip: PythonEnvironmentSpec =
+            serde_json::from_str(r#"{"kind":"pip","packages":["numpy"]}"#).unwrap();
+        assert!(!serde_json::to_string(&pip).unwrap().contains("channels"));
+    }
+}
@@ -63,4 +63,5 @@ pub struct FmIndexBuilder {}
 pub use lance_index::scalar::FullTextSearchQuery;
 pub use lance_index::scalar::InvertedIndexParams as FtsIndexBuilder;
 pub use lance_index::scalar::InvertedIndexParams;
+pub use lance_index::scalar::inverted::DocumentGranularity;
 pub use lance_index::scalar::inverted::query::*;
@@ -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 {
@@ -87,6 +87,10 @@ impl ServerVersion {
    pub fn support_blobs(&self) -> bool {
        self.0 >= semver::Version::new(0, 5, 0)
    }
+
+    pub fn support_fts_document_granularity(&self) -> bool {
+        self.0 >= semver::Version::new(0, 6, 0)
+    }
 }

 pub const OPT_REMOTE_PREFIX: &str = "remote_database_";
@@ -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))
            }
@@ -59,7 +59,9 @@ use crate::index::{IndexConfig, IndexStatisticsImpl, IndexType};
 use crate::job::Job;
 use crate::query::{IntoQueryVector, Query, QueryExecutionOptions, TakeQuery, VectorQuery};
 use crate::table::datafusion::insert::InsertExec;
-use crate::utils::{PatchReadParam, PatchWriteParam, resolve_arrow_field_path};
+use crate::utils::{
+    PatchReadParam, PatchWriteParam, public_fts_field_path_by_id, resolve_arrow_field_path,
+};

 use self::dataset::DatasetConsistencyWrapper;
 use self::merge::MergeInsertBuilder;
@@ -560,6 +562,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.
@@ -785,6 +794,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 +1148,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
@@ -1752,23 +1777,7 @@ impl Table {
        self.inner.alter_columns(alterations).await
    }

-    /// Update per-field (column) metadata.
-    ///
-    /// Each [`FieldMetadataUpdate`] is merged into the field's existing metadata
-    /// by default; use [`FieldMetadataUpdate::remove`] to delete a key, or
-    /// [`FieldMetadataUpdate::replace`] to swap the field's entire metadata map.
-    ///
-    /// The following keys are treated specially, by convention, and should be
-    /// used when appropriate:
-    ///
-    /// - `lancedb:description`: for a human-readable description of a field.
-    /// - `lancedb:tag:<name>`: for a user-defined key-value tag, where the suffix
-    ///   names the tag category; e.g. `lancedb:tag:model: "clip"`.
-    /// - `lancedb:logical-column`: for a column grouping; e.g. `feature_v1` and
-    ///   `feature_v2` might be in the same logical column.
-    /// - `lancedb:status`: for status options (`production`, `candidate`,
-    ///   `deprecated`, `archived`) to designate the current life cycle state of
-    ///   this column.
+    /// Update per-field metadata (merges by default).
    pub async fn update_field_metadata(
        &self,
        updates: &[FieldMetadataUpdate],
@@ -1960,6 +1969,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.
@@ -3055,10 +3078,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
    }
@@ -3515,7 +3561,14 @@ impl BaseTable for NativeTable {
                let field_ids = idx_desc.field_ids();
                let mut columns = Vec::with_capacity(field_ids.len());
                for field_id in field_ids {
-                    let field_path = match dataset.schema().field_path(*field_id as i32) {
+                    let field_path = match if index_type == crate::index::IndexType::FTS {
+                        public_fts_field_path_by_id(dataset.schema(), *field_id as i32)
+                    } else {
+                        dataset
+                            .schema()
+                            .field_path(*field_id as i32)
+                            .map_err(Into::into)
+                    } {
                        Ok(field_path) => field_path,
                        Err(e) => {
                            log::warn!(
@@ -4102,6 +4155,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]
@@ -4205,6 +4266,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]
@@ -28,8 +28,9 @@ pub(super) type PreparedIndex = (String, Box<dyn lance::index::IndexParams>, Ind
 use crate::index::Index;
 use crate::index::vector::{VectorIndex, suggested_num_sub_vectors};
 use crate::utils::{
-    supported_bitmap_data_type, supported_btree_data_type, supported_fm_data_type,
-    supported_fts_data_type, supported_label_list_data_type, supported_vector_data_type,
+    resolve_lance_fts_field_path, supported_bitmap_data_type, supported_btree_data_type,
+    supported_fm_data_type, supported_fts_data_type, supported_label_list_data_type,
+    supported_vector_data_type,
 };

 use super::NativeTable;
@@ -122,7 +123,20 @@ impl NativeTable {
        }
        self.dataset.ensure_mutable()?;
        let dataset = self.dataset.get().await?;
-        let (column, field) = Self::resolve_index_field(dataset.schema(), &opts.columns[0])?;
+        let (column, field) = if let Index::FTS(params) = &opts.index {
+            let resolved = resolve_lance_fts_field_path(dataset.schema(), &opts.columns[0])?;
+            if params.get_document_granularity().is_list_element() && resolved.list_depth == 0 {
+                return Err(Error::InvalidInput {
+                    message: format!(
+                        "FTS field path '{}' has no List layer and cannot use ListElement document granularity",
+                        resolved.canonical_path
+                    ),
+                });
+            }
+            (resolved.canonical_path, resolved.field)
+        } else {
+            Self::resolve_index_field(dataset.schema(), &opts.columns[0])?
+        };
        let params = self.make_index_params(&field, opts.index.clone()).await?;
        let index_type = self.get_index_type_for_field(&field, &opts.index);
        Ok((column, params, index_type))
@@ -436,7 +450,7 @@ mod tests {
    use crate::connection::ConnectBuilder;
    use crate::index::Index;
    use crate::index::scalar::{
-        BTreeIndexBuilder, BitmapIndexBuilder, FmIndexBuilder, FtsIndexBuilder,
+        BTreeIndexBuilder, BitmapIndexBuilder, DocumentGranularity, FmIndexBuilder, FtsIndexBuilder,
    };
    use crate::index::vector::{
        IvfHnswFlatIndexBuilder, IvfHnswPqIndexBuilder, IvfHnswSqIndexBuilder,
@@ -553,6 +567,38 @@ mod tests {
        job.cancel().await.unwrap();
    }

+    #[tokio::test]
+    async fn test_execute_async_validates_fts_input_before_starting_job() {
+        let conn = connect("memory://").execute().await.unwrap();
+        let batch =
+            record_batch!(("id", Int32, [1, 2]), ("text", Utf8, ["alpha", "beta"])).unwrap();
+        let table = conn.create_table("t", batch).execute().await.unwrap();
+
+        let missing = table
+            .create_index(&["missing"], Index::FTS(FtsIndexBuilder::default()))
+            .execute_async()
+            .await;
+        assert!(missing.is_err());
+
+        let invalid_type = table
+            .create_index(&["id"], Index::FTS(FtsIndexBuilder::default()))
+            .execute_async()
+            .await;
+        assert!(invalid_type.is_err());
+
+        let invalid_granularity = table
+            .create_index(
+                &["text"],
+                Index::FTS(
+                    FtsIndexBuilder::default()
+                        .document_granularity(DocumentGranularity::ListElement),
+                ),
+            )
+            .execute_async()
+            .await;
+        assert!(invalid_granularity.is_err());
+    }
+
    /// Concurrent waiters, and a wait issued after the job settled, all
    /// succeed once the build does.
    #[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();
@@ -697,6 +697,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 +890,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 +1056,37 @@ mod tests {
        assert_eq!(namespace_client.query_table_calls.load(Ordering::SeqCst), 0);
    }

+    #[tokio::test]
+    async fn test_query_snapshot_disables_namespace_pushdown() {
+        use crate::connect;
+        use crate::table::BaseTable;
+        use arrow_array::{Int32Array, RecordBatch};
+        use arrow_schema::{DataType, Field, Schema};
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        let batch =
+            RecordBatch::try_new(schema, vec![Arc::new(Int32Array::from(vec![1, 2, 3]))]).unwrap();
+        let table = conn
+            .create_table("test_snapshot_namespace_fallback", vec![batch])
+            .execute()
+            .await
+            .unwrap();
+        let mut native_table = table.as_native().unwrap().clone();
+        native_table.namespace_client = Some(Arc::new(CountingNamespaceClient::default()));
+        native_table
+            .pushdown_operations
+            .insert(NamespaceClientPushdownOperation::QueryTable);
+
+        let snapshot = BaseTable::query_snapshot(&native_table).await.unwrap();
+        let snapshot = snapshot.as_any().downcast_ref::<NativeTable>().unwrap();
+        assert!(
+            !can_execute_namespace_query(snapshot, &AnyQuery::Query(QueryRequest::default()),)
+                .await
+                .unwrap()
+        );
+    }
+
    #[tokio::test]
    async fn test_create_plan_multivector_structure() {
        use arrow_array::{Float32Array, RecordBatch};
@@ -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));
        }
    }
@@ -55,9 +55,7 @@ pub struct DropColumnsResult {
 pub struct FieldMetadataUpdate {
    /// Dot-separated path to the field (e.g. `"embedding"` or `"address.zip"`).
    pub path: String,
-    /// Keys to set (`Some`) or delete (`None`). See
-    /// [`Table::update_field_metadata`](crate::Table::update_field_metadata) for
-    /// the conventional `lancedb:*` keys.
+    /// Keys to set (`Some`) or delete (`None`).
    pub metadata: HashMap<String, Option<String>>,
    /// If `true`, replace the field's entire metadata map instead of merging.
    pub replace: bool,
@@ -225,6 +225,159 @@ pub(crate) fn resolve_arrow_field_path(schema: &Schema, column: &str) -> Result<
    Ok((canonical_path, Field::from(*field)))
 }

+pub(crate) struct ResolvedFtsField {
+    pub canonical_path: String,
+    pub field: Field,
+    pub list_depth: usize,
+}
+
+/// Canonicalize a public FTS field path while keeping Arrow list item names hidden.
+pub(crate) fn resolve_lance_fts_field_path(
+    schema: &lance_core::datatypes::Schema,
+    column: &str,
+) -> Result<ResolvedFtsField> {
+    let names =
+        lance_core::datatypes::parse_field_path(column).map_err(|e| Error::InvalidInput {
+            message: format!("Invalid field path `{}`: {}", column, e),
+        })?;
+    let (root_name, remaining_names) = names.split_first().ok_or_else(|| Error::InvalidInput {
+        message: "FTS field path cannot be empty".to_string(),
+    })?;
+    let mut field = schema
+        .fields
+        .iter()
+        .find(|field| field.name == *root_name)
+        .or_else(|| {
+            schema
+                .fields
+                .iter()
+                .find(|field| field.name.eq_ignore_ascii_case(root_name))
+        })
+        .ok_or_else(|| fts_field_not_found(schema, column))?;
+    let mut canonical_names = vec![field.name.clone()];
+    let mut list_depth = 0;
+
+    for name in remaining_names {
+        while matches!(
+            field.data_type(),
+            DataType::List(_) | DataType::LargeList(_)
+        ) {
+            list_depth += 1;
+            field = field.children.first().ok_or_else(|| Error::Schema {
+                message: format!(
+                    "FTS field path `{}` has a list without an item field",
+                    column
+                ),
+            })?;
+        }
+        if !matches!(field.data_type(), DataType::Struct(_)) {
+            return Err(fts_field_not_found(schema, column));
+        }
+        field = field
+            .children
+            .iter()
+            .find(|field| field.name == *name)
+            .or_else(|| {
+                field
+                    .children
+                    .iter()
+                    .find(|field| field.name.eq_ignore_ascii_case(name))
+            })
+            .ok_or_else(|| fts_field_not_found(schema, column))?;
+        canonical_names.push(field.name.clone());
+    }
+
+    let mut terminal = field;
+    while matches!(
+        terminal.data_type(),
+        DataType::List(_) | DataType::LargeList(_)
+    ) {
+        list_depth += 1;
+        terminal = terminal.children.first().ok_or_else(|| Error::Schema {
+            message: format!(
+                "FTS field path `{}` has a list without an item field",
+                column
+            ),
+        })?;
+    }
+
+    let canonical_path = lance_core::datatypes::format_field_path(
+        &canonical_names
+            .iter()
+            .map(String::as_str)
+            .collect::<Vec<_>>(),
+    );
+    Ok(ResolvedFtsField {
+        canonical_path,
+        field: Field::from(field),
+        list_depth,
+    })
+}
+
+fn fts_field_not_found(schema: &lance_core::datatypes::Schema, column: &str) -> Error {
+    Error::Schema {
+        message: format!(
+            "Field path `{}` not found in schema. Available field paths: {}",
+            column,
+            schema.field_paths().join(", ")
+        ),
+    }
+}
+
+fn find_public_fts_field_path_by_id(
+    field: &lance_core::datatypes::Field,
+    field_id: i32,
+    path: &mut Vec<String>,
+) -> bool {
+    if field.id == field_id {
+        return true;
+    }
+    match field.data_type() {
+        DataType::List(_) | DataType::LargeList(_) => field
+            .children
+            .first()
+            .is_some_and(|child| find_public_fts_field_path_by_id(child, field_id, path)),
+        DataType::Struct(_) => field.children.iter().any(|child| {
+            path.push(child.name.clone());
+            let found = find_public_fts_field_path_by_id(child, field_id, path);
+            if !found {
+                path.pop();
+            }
+            found
+        }),
+        _ => false,
+    }
+}
+
+pub(crate) fn public_fts_field_path_by_id(
+    schema: &lance_core::datatypes::Schema,
+    field_id: i32,
+) -> Result<String> {
+    for root in &schema.fields {
+        let mut path = vec![root.name.clone()];
+        if find_public_fts_field_path_by_id(root, field_id, &mut path) {
+            return Ok(lance_core::datatypes::format_field_path(
+                &path.iter().map(String::as_str).collect::<Vec<_>>(),
+            ));
+        }
+    }
+    Err(Error::Schema {
+        message: format!("Field id `{}` not found in schema", field_id),
+    })
+}
+
+pub(crate) fn resolve_arrow_fts_field_path(
+    schema: &Schema,
+    column: &str,
+) -> Result<(String, Field)> {
+    let lance_schema =
+        lance_core::datatypes::Schema::try_from(schema).map_err(|e| Error::Schema {
+            message: format!("Invalid schema: {}", e),
+        })?;
+    let resolved = resolve_lance_fts_field_path(&lance_schema, column)?;
+    Ok((resolved.canonical_path, resolved.field))
+}
+
 pub fn supported_btree_data_type(dtype: &DataType) -> bool {
    dtype.is_integer()
        || dtype.is_floating()
@@ -480,6 +633,36 @@ mod tests {

    use super::*;

+    #[test]
+    fn test_public_fts_field_path_prefers_exact_case() {
+        let text_list = || {
+            DataType::List(Arc::new(Field::new(
+                "item",
+                DataType::Struct(vec![Field::new("content", DataType::Utf8, true)].into()),
+                true,
+            )))
+        };
+        let schema = Schema::new(vec![
+            Field::new("Docs", text_list(), true),
+            Field::new("docs", text_list(), true),
+        ]);
+
+        let (path, _) = resolve_arrow_fts_field_path(&schema, "docs.content").unwrap();
+        assert_eq!(path, "docs.content");
+
+        let lance_schema = lance_core::datatypes::Schema::try_from(&schema).unwrap();
+        let field_id = lance_schema
+            .resolve_case_insensitive("docs.item.content")
+            .unwrap()
+            .last()
+            .unwrap()
+            .id;
+        assert_eq!(
+            public_fts_field_path_by_id(&lance_schema, field_id).unwrap(),
+            "docs.content"
+        );
+    }
+
    #[test]
    fn test_guess_default_column() {
        let schema_no_vector = Schema::new(vec![
Author	SHA1	Message	Date
Xuanwo	b78f2a5044	feat: expose list-element FTS document granularity (#4050 ) ## Summary LanceDB could not request Lance's list-element FTS document granularity through Python or Remote APIs, and generic nested-field resolution exposed Arrow's internal `item` segment instead of the public field path. This exposes typed `row \| list_element` configuration for Python FTS index creation and match/phrase queries, preserves `_doc_index`, and keeps nested FTS paths public (for example, `docs.content`). Remote list-element requests require server API version 0.6.0 so older servers cannot silently execute them with row semantics; explicit row requests remain compatible. ## Compatibility Omitted index and query parameters retain row granularity. Remote row index creation omits the new wire field. ## Tracking [ENT-2342](https://linear.app/lancedb/issue/ENT-2342/expose-list-element-fts-document-granularity-end-to-end)	2026-08-26 23:27:39 +08:00
Wyatt Alt	06872463cf	feat: declare conda environments on Functions (#4057 ) A Function's remote environment can now be conda instead of pip. `@udf(conda=[...], conda_channels=[...])` registers one; pip and conda are exclusive, channels are priority-ordered and require conda. The Rust and Python `PythonEnvironmentSpec` models gain `channels`, dropped from the canonical JSON when empty so existing pip registrations keep their digests.	2026-08-26 06:29:43 -07:00
lancedb-gatefixer[bot]	2fbf6d6211	test(python): cover concurrent S3 table opens (#3833 ) ## Summary - add regression coverage for the reported synchronous Python workload with 32 simultaneous `open_table` calls - verify every independently opened S3-backed table handle can read through the connection's shared session and object-store client ## Root cause In Python v0.13.0, each synchronous table handle lazily constructed its own Lance dataset. Opening many handles in parallel therefore triggered independent S3 client construction and bucket-region resolution, which failed under thread pressure. The current Rust-backed connection path owns a shared Lance session and retains its object-store handle, so table opens reuse the existing S3 client; these tests lock in that behavior through the public Python API and a causal Session-registry invariant. ## Validation - `uvx --from 'ruff==0.15.20' ruff format --check python/tests/test_s3.py` - `uvx --from 'ruff==0.15.20' ruff check .` - `cargo fmt --all` - `cargo test --quiet --features remote -p lancedb test_concurrent_open_table_reuses_connection_object_store` - `cargo check --quiet --features remote --tests --examples` - equivalent 32-thread `open_table(...).count_rows()` workload against a local database - targeted S3 test collected successfully locally; execution requires the CI LocalStack service, which is unavailable in this runner Fixes #1786 <!-- lance-gatekeeper-fix:v1 agent=d311f3c7151f77ae22b4997702e7b7db generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-26 14:56:57 +08: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