fix(node): accept Arrow metadata across JavaScript realms (#3904 )

## Summary - accept genuine Arrow metadata maps created in another JavaScript realm - validate every metadata entry and clone it into a local Map - cover an Arrow 15 VM-realm table through the public fromDataToBuffer boundary - retain structural typing for nested and dictionary Arrow data ## Root cause The sanitizer used a local-realm instanceof Map check for schema and field metadata. A genuine Map created in another JavaScript realm has the required internal Map state but fails that identity check, so fromDataToBuffer rejected the foreign table before serializing its rows. ## Scope This fixes the distinct JavaScript-realm sanitizer failure identified during review. It does not establish the cause of the S3/compaction panic reported in #1525, so that issue remains open. ## Validation - pnpm test --runInBand (707 passed, 5 skipped) - pnpm test --runInBand __test__/arrow.test.ts (189 passed) - pnpm build - pnpm lint - pnpm run docs Related to #1525  --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>
feat(python): accept expressions in update filters (#3876 )
2026-08-27 16:38:31 +00:00 · 2026-08-27 23:35:02 +08:00 · 2026-08-27 20:28:21 +08:00 · 2026-08-27 17:10:51 +08:00 · 2026-08-27 15:47:18 +08:00 · 2026-08-27 13:42:30 +08:00
75 changed files with 4772 additions and 777 deletions
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.38.0-beta.10"
+current_version = "0.38.0-beta.11"
 parse = """(?x)
    (?P<major>0|[1-9]\\d*)\\.
    (?P<minor>0|[1-9]\\d*)\\.
@@ -3455,8 +3455,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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 = "12.0.0-beta.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v12.0.0-beta.1#e37cee4397abc6a71df3cdfa0e637c274f820aa6"
+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",
@@ -5402,7 +5402,7 @@ dependencies = [

 [[package]]
 name = "lancedb"
-version = "0.38.0-beta.10"
+version = "0.38.0-beta.11"
 dependencies = [
 "ahash",
 "anyhow",
@@ -5490,7 +5490,7 @@ dependencies = [

 [[package]]
 name = "lancedb-nodejs"
-version = "0.38.0-beta.10"
+version = "0.38.0-beta.11"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5515,7 +5515,7 @@ dependencies = [

 [[package]]
 name = "lancedb-python"
-version = "0.38.0-beta.10"
+version = "0.38.0-beta.11"
 dependencies = [
 "arrow",
 "async-trait",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=12.0.0-beta.1", default-features = false, "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=12.0.0-beta.1", default-features = false, "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=12.0.0-beta.1", default-features = false, "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=12.0.0-beta.1", "tag" = "v12.0.0-beta.1", "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
@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-core</artifactId>
-    <version>0.38.0-beta.10</version>
+    <version>0.38.0-beta.11</version>
 </dependency>
 ```

@@ -1292,6 +1292,18 @@ 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)[]
@@ -17,7 +17,8 @@ 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.
+default; a value of `null` deletes that key. See
+[Table.updateFieldMetadata](../classes/Table.md#updatefieldmetadata) for the conventional `lancedb:*` keys.

 ***

@@ -159,6 +159,8 @@ and combined with [BooleanQuery][lancedb.query.BooleanQuery].

 ::: lancedb.query.FullTextOperator

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

 ## Embeddings
@@ -8,7 +8,7 @@
    <parent>
      <groupId>com.lancedb</groupId>
      <artifactId>lancedb-parent</artifactId>
-      <version>0.38.0-beta.10</version>
+      <version>0.38.0-beta.11</version>
      <relativePath>../pom.xml</relativePath>
    </parent>

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

    <groupId>com.lancedb</groupId>
    <artifactId>lancedb-parent</artifactId>
-    <version>0.38.0-beta.10</version>
+    <version>0.38.0-beta.11</version>
    <packaging>pom</packaging>
    <name>${project.artifactId}</name>
    <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>12.0.0-beta.1</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>
@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.38.0-beta.10"
+version = "0.38.0-beta.11"
 publish = false
 license.workspace = true
 description.workspace = true
@@ -1,11 +1,16 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
+import * as fs from "node:fs";
+import * as vm from "node:vm";
 import * as arrow15 from "apache-arrow-15";
 import * as arrow16 from "apache-arrow-16";
 import * as arrow17 from "apache-arrow-17";
 import * as arrow18 from "apache-arrow-18";

 import {
+  Field as CurrentField,
+  LargeBinary as CurrentLargeBinary,
+  Schema as CurrentSchema,
  Vector as CurrentVector,
  convertToTable,
  tableFromIPC as currentTableFromIPC,
@@ -36,6 +41,59 @@ function sampleRecords(): Array<Record<string, any>> {
    },
  ];
 }
+
+it("serializes an Arrow Table created in another JavaScript realm", async () => {
+  const context = vm.createContext({
+    TextDecoder,
+    TextEncoder,
+    console,
+    setTimeout,
+    clearTimeout,
+  });
+  vm.runInContext(
+    fs.readFileSync(
+      require.resolve("apache-arrow-15/Arrow.es2015.min"),
+      "utf8",
+    ),
+    context,
+  );
+  const foreignTable: unknown = vm.runInContext(
+    "Arrow.tableFromArrays({ id: new Int32Array([1, 2, 3]), text: ['foo', 'bar', 'baz'] })",
+    context,
+  );
+
+  const foreignMetadata = (
+    foreignTable as { schema: { metadata: Map<string, string> } }
+  ).schema.metadata;
+  expect(foreignMetadata).not.toBeInstanceOf(Map);
+
+  const buf = await fromDataToBuffer(
+    foreignTable as Parameters<typeof fromDataToBuffer>[0],
+  );
+  const actual = currentTableFromIPC(buf);
+
+  expect(actual.numRows).toBe(3);
+  expect(actual.getChild("id")?.toJSON()).toEqual([1, 2, 3]);
+  expect(actual.getChild("text")?.toJSON()).toEqual(["foo", "bar", "baz"]);
+});
+
+it("preserves field metadata from a provided schema", async function () {
+  const jsonMetadata = new Map([["ARROW:extension:name", "lance.json"]]);
+  const schema = new CurrentSchema([
+    new CurrentField("meta", new CurrentLargeBinary(), true, jsonMetadata),
+  ]);
+
+  const table = makeArrowTable(
+    [{ meta: Buffer.from(JSON.stringify({ source: "test" })) }],
+    { schema },
+  );
+
+  expect(table.schema.fields[0].metadata).toEqual(jsonMetadata);
+
+  const roundTripped = currentTableFromIPC(await fromTableToBuffer(table));
+  expect(roundTripped.schema.fields[0].metadata).toEqual(jsonMetadata);
+});
+
 describe.each([arrow15, arrow16, arrow17, arrow18])(
  "Arrow",
  (
@@ -187,6 +187,58 @@ describe("embedding functions", () => {
    const vector0 = JSON.parse(JSON.stringify(arr[0].vector));
    expect(vector0).toEqual([1, 2, 3]);
  });
+  it("should append multiple Python embeddings with the same alias", async () => {
+    @register("python-mock")
+    // biome-ignore lint/correctness/noUnusedVariables: the decorator registers this class
+    class MockEmbeddingFunction extends EmbeddingFunction<string> {
+      ndims() {
+        return 3;
+      }
+      embeddingDataType(): Float {
+        return new Float32();
+      }
+      async computeQueryEmbeddings(_data: string) {
+        return [1, 2, 3];
+      }
+      async computeSourceEmbeddings(data: string[]) {
+        return data.map((value) =>
+          value === "hello world" ? [1, 2, 3] : [4, 5, 6],
+        );
+      }
+    }
+
+    const metadata = new Map([
+      [
+        "embedding_functions",
+        '[{"source_column":"text1","vector_column":"vector1","name":"python-mock","model":{}},{"source_column":"text2","vector_column":"vector2","name":"python-mock","model":{}}]',
+      ],
+    ]);
+    const schema = new Schema(
+      [
+        new Field("text1", new Utf8(), true),
+        new Field("text2", new Utf8(), true),
+        new Field(
+          "vector1",
+          new FixedSizeList(3, new Field("item", new Float32(), true)),
+          true,
+        ),
+        new Field(
+          "vector2",
+          new FixedSizeList(3, new Field("item", new Float32(), true)),
+          true,
+        ),
+      ],
+      metadata,
+    );
+
+    const db = await connect(tmpDir.name);
+    const table = await db.createEmptyTable("test", schema);
+    await table.add([{ text1: "hello world", text2: "goodbye world" }]);
+
+    const rows = await table.query().toArray();
+    expect(JSON.parse(JSON.stringify(rows[0].vector1))).toEqual([1, 2, 3]);
+    expect(JSON.parse(JSON.stringify(rows[0].vector2))).toEqual([4, 5, 6]);
+  });

  it("should append generated vectors to a non-nullable schema", async () => {
    @register("non_nullable_schema_test")
@@ -2585,7 +2585,24 @@ describe.each([arrow15, arrow16, arrow17, arrow18])(
      );
    });

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

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

+  test("observes promised additional vectors while the query is pending", async () => {
+    const initialVector = new Promise<number[]>(() => undefined);
+    const query = table.query().nearestTo(initialVector);
+    const unhandled: unknown[] = [];
+    const onUnhandled = (reason: unknown) => unhandled.push(reason);
+    process.on("unhandledRejection", onUnhandled);
+
+    try {
+      query.addQueryVector(Promise.reject(new Error("extra vector failed")));
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(unhandled).toEqual([]);
+
+      const rejectedQuery = table
+        .query()
+        .nearestTo([0.1, 0.2])
+        .addQueryVector(Promise.reject(new Error("consumed vector failed")));
+      await expect(rejectedQuery.toArray()).rejects.toThrow(
+        "consumed vector failed",
+      );
+    } finally {
+      process.off("unhandledRejection", onUnhandled);
+    }
+  });
+
  test("index and search multivectors", async () => {
    const db = await connect(tmpDir.name);
    const data = [];
@@ -3220,6 +3561,27 @@ describe("when creating an empty table", () => {
    expect((actualSchema.fields[1].type as Float64).precision).toBe(2);
  });

+  it("can add and query JSON data", async () => {
+    const schema = new Schema([
+      new Field("id", new Int32(), true),
+      new Field(
+        "meta",
+        new Utf8(),
+        true,
+        new Map([["ARROW:extension:name", "arrow.json"]]),
+      ),
+    ]);
+    const table = await con.createEmptyTable("json", schema);
+    const meta = JSON.stringify({ x: 1 });
+
+    await table.add([{ id: 1, meta }]);
+
+    const rows = await table.query().toArray();
+    expect(rows).toHaveLength(1);
+    expect(rows[0].id).toBe(1);
+    expect(rows[0].meta).toBe(meta);
+  });
+
  it("can create an empty table from schema that specifies field types by name", async () => {
    const schemaLike = {
      fields: [
@@ -170,7 +170,7 @@ test("basic table examples", async () => {
    // --8<-- [end:create_index]

    // --8<-- [start:delete_rows]
-    await tbl.delete('item = "fizz"');
+    await tbl.delete("item = 'fizz'");
    // --8<-- [end:delete_rows]

    // --8<-- [start:drop_table]
@@ -72,8 +72,7 @@ export type FieldLike =
    };

 export type DataLike =
-  // biome-ignore lint/suspicious/noExplicitAny: <explanation>
-  | import("apache-arrow").Data<Struct<any>>
+  | import("apache-arrow").Data
  | {
      // biome-ignore lint/suspicious/noExplicitAny: <explanation>
      type: any;
@@ -82,6 +81,7 @@ export type DataLike =
      stride: number;
      nullable: boolean;
      children: DataLike[];
+      dictionary?: { data: readonly DataLike[] };
      get nullCount(): number;
      // biome-ignore lint/suspicious/noExplicitAny: <explanation>
      values: Buffers<any>[BufferType.DATA];
@@ -100,6 +100,29 @@ export interface FullTextSearchOptions {
  columns?: string | string[];
 }

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

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

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

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

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

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

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

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

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

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

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

@@ -697,43 +727,39 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
   * Add a query vector to the search
   *
   * This method can be called multiple times to add multiple query vectors
-   * to the search. If multiple query vectors are added, then they will be searched
-   * in parallel, and the results will be concatenated. A column called `query_index`
-   * will be added to indicate the index of the query vector that produced the result.
-   *
-   * Performance wise, this is equivalent to running multiple queries concurrently.
+   * to the search. A column called `query_index` will be added to indicate the index
+   * of the query vector that produced the result. Flat searches share one table scan
+   * across the query vectors, avoiding the scan and memory amplification of running
+   * multiple queries concurrently. Indexed searches may still perform per-vector
+   * index work.
   */
  addQueryVector(vector: IntoVector): VectorQuery {
    if (vector instanceof Promise) {
+      // Observe the promise as soon as it is accepted. The existing native
+      // query may still be pending, and delaying observation until it resolves
+      // can otherwise surface a fast rejection as unhandled.
+      const settledVector = vector.then(
+        (value) => ({ status: "fulfilled" as const, value }),
+        (reason) => ({ status: "rejected" as const, reason }),
+      );
      const res = (async () => {
-        try {
-          const v = await vector;
-          // biome-ignore lint/suspicious/noExplicitAny: we need to get the `inner`, but js has no package scoping
-          const value: any = this.addQueryVector(v);
-          const inner = value.inner as
-            | NativeVectorQuery
-            | Promise<NativeVectorQuery>;
-          return inner;
-        } catch (e) {
-          return Promise.reject(e);
+        const inner = await this.getInner();
+        const outcome = await settledVector;
+        if (outcome.status === "rejected") {
+          throw outcome.reason;
        }
+        addQueryVectorToNative(inner, outcome.value);
+        return inner;
      })();
      return new VectorQuery(res);
    } else {
-      super.doCall((inner) => {
-        const raw = Array.isArray(vector) ? null : extractVectorBuffer(vector);
-        if (raw) {
-          inner.addQueryVectorRaw(raw.data, raw.dtype);
-        } else {
-          inner.addQueryVector(Float32Array.from(vector as number[]));
-        }
-      });
+      this.doVectorCall((inner) => addQueryVectorToNative(inner, vector));
      return this;
    }
  }

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

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

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

  nearestToText(query: string | FullTextQuery, columns?: string[]): Query {
@@ -94,17 +94,24 @@ export function sanitizeMetadata(
  if (metadataLike === undefined || metadataLike === null) {
    return undefined;
  }
-  if (!(metadataLike instanceof Map)) {
+
+  let entries: IterableIterator<[unknown, unknown]>;
+  try {
+    entries = Map.prototype.entries.call(metadataLike);
+  } catch {
    throw Error("Expected metadata, if present, to be a Map<string, string>");
  }
-  for (const item of metadataLike) {
-    if (typeof item[0] !== "string" || typeof item[1] !== "string") {
+
+  const metadata = new Map<string, string>();
+  for (const [key, value] of entries) {
+    if (typeof key !== "string" || typeof value !== "string") {
      throw Error(
        "Expected metadata, if present, to be a Map<string, string> but it had non-string keys or values",
      );
    }
+    metadata.set(key, value);
  }
-  return metadataLike as Map<string, string>;
+  return metadata;
 }

 export function sanitizeInt(typeLike: object) {
@@ -406,10 +406,11 @@ function matchingFields(fields: Field[], tree: FieldTree): Field[] {
          field.name,
          new Struct(matchingFields(struct.children, value)),
          field.nullable,
+          field.metadata,
        ),
      );
    } else {
-      matches.push(new Field(field.name, value as DataType, field.nullable));
+      matches.push(field);
    }
  }
  return matches;
@@ -48,6 +48,7 @@ import {
  Query,
  TakeQuery,
  VectorQuery,
+  createAutoQuery,
  instanceOfFullTextQuery,
 } from "./query";
 import { sanitizeType } from "./sanitize";
@@ -629,6 +630,18 @@ 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.
@@ -1177,33 +1190,29 @@ export class LocalTable extends Table {
      });
    }

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

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

    const queryPromise = this.getEmbeddingFunctions().then(
@@ -1558,7 +1567,8 @@ 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.
+   * default; a value of `null` deletes that key. See
+   * {@link Table.updateFieldMetadata} for the conventional `lancedb:*` keys.
   */
  metadata: Record<string, string | null>;
  /** If true, replace the field's entire metadata map instead of merging. */
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.38.0-beta.10",
+	"version": "0.38.0-beta.11",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.38.0-beta.10",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.38.0-beta.10",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.38.0-beta.10",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.38.0-beta.10",
+	"version": "0.38.0-beta.11",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.38.0-beta.10",
+  "version": "0.38.0-beta.11",
  "os": [
    "win32"
  ],
@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.38.0-beta.10",
+	"version": "0.38.0-beta.11",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",
@@ -1,12 +1,12 @@
 {
  "name": "@lancedb/lancedb",
-  "version": "0.38.0-beta.10",
+  "version": "0.38.0-beta.11",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "@lancedb/lancedb",
-      "version": "0.38.0-beta.10",
+      "version": "0.38.0-beta.11",
      "cpu": [
        "x64",
        "arm64"
@@ -11,7 +11,7 @@
    "ann"
  ],
  "private": false,
-  "version": "0.38.0-beta.10",
+  "version": "0.38.0-beta.11",
  "main": "dist/index.js",
  "exports": {
    ".": "./dist/index.js",
@@ -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(
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.38.0-beta.10"
+version = "0.38.0-beta.11"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"
@@ -270,7 +270,8 @@ def _iter_projection_pairs(
            if isinstance(expr, str):
                yield name, expr
            elif isinstance(expr, Expr):
-                yield name, expr.to_sql()
+                source = expr._column_name()
+                yield name, source if source is not None else expr.to_sql()
        return
    for column in projection:
        if isinstance(column, str):
@@ -280,7 +281,8 @@ def _iter_projection_pairs(
            if isinstance(expr, str):
                yield name, expr
            elif isinstance(expr, Expr):
-                yield name, expr.to_sql()
+                source = expr._column_name()
+                yield name, source if source is not None else expr.to_sql()


 def _set_blob_column(tbl: pa.Table, output_name: str, blobs: pa.Array) -> pa.Table:
@@ -87,6 +87,7 @@ class PyExpr:
    def contains(self, substr: "PyExpr") -> "PyExpr": ...
    def isin(self, values: List["PyExpr"]) -> "PyExpr": ...
    def cast(self, data_type: pa.DataType) -> "PyExpr": ...
+    def column_name(self) -> Optional[str]: ...
    def to_sql(self) -> str: ...

 def expr_col(name: str) -> PyExpr: ...
@@ -608,6 +609,7 @@ class PyQueryRequest:
    filter: Optional[Union[str, bytes]]
    full_text_search: Optional[FullTextQuery]
    select: Optional[Union[str, List[str]]]
+    select_source_columns: Optional[Dict[str, str]]
    fast_search: Optional[bool]
    with_row_id: Optional[bool]
    use_lsm: Optional[bool]
@@ -249,6 +249,10 @@ class Expr:

    # ── utilities ────────────────────────────────────────────────────────────

+    def _column_name(self) -> str | None:
+        """Return the source name when this is a bare column expression."""
+        return self._inner.column_name()
+
    def to_sql(self) -> str:
        """Render the expression as a SQL string (useful for debugging)."""
        return self._inner.to_sql()
@@ -312,7 +316,7 @@ def func(name: str, *args: ExprLike) -> Expr:
    --------
    >>> from lancedb.expr import col, func
    >>> func("lower", col("name"))
-    Expr(lower(name))
+    Expr(lower(`name`))
    """
    inner_args = [_coerce(a)._inner for a in args]
    return Expr(expr_func(name, inner_args))
@@ -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
@@ -167,6 +167,12 @@ def _projection_to_scanner_kwargs(columns: QueryProjection) -> Dict[str, Any]:
    return {"columns": projection}


+def _query_request_projection(req: "PyQueryRequest") -> QueryProjection:
+    if req.select_source_columns is not None:
+        return req.select_source_columns
+    return req.select
+
+
 def _scanner_kwargs_for_query(
    query: Query,
    blob_mode: BlobMode,
@@ -375,6 +381,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 +491,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 +504,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 +523,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
@@ -2776,15 +2805,16 @@ class AsyncQueryBase(object):

        req = self._inner.to_query_request()
        schema = await self._table.schema()
+        projection = _query_request_projection(req)
        self._blob_auto_row_id = blob_auto_row_id_for_scan(
            schema,
-            req.select,
+            projection,
            with_row_id=self._with_row_id,
        )
        if not self._blob_auto_row_id:
            self._blob_paths = ()
            return
-        self._blob_paths = tuple(blob_v2_projection_sources(schema, req.select).keys())
+        self._blob_paths = tuple(blob_v2_projection_sources(schema, projection).keys())
        self._inner.with_row_id()

    def select(self, columns: Union[List[str], dict[str, str]]) -> Self:
@@ -3378,9 +3408,10 @@ class AsyncQuery(AsyncStandardQuery):
        pass in multiple vectors. When multiple vectors are passed in, if the vector
        column is with multivector type, then the vectors will be treated as a single
        query. Or the vectors will be treated as multiple queries, this can be useful
-        if you want to find the nearest vectors to multiple query vectors.
-        This is not expected to be faster than making multiple queries concurrently;
-        it is just a convenience method. If multiple vectors are passed in then
+        if you want to find the nearest vectors to multiple query vectors. Flat
+        searches share one table scan across the query vectors, avoiding the scan
+        and memory amplification of making multiple queries concurrently. If
+        multiple vectors are passed in then
        an additional column `query_index` will be added to the results. This column
        will contain the index of the query vector that the result is nearest to.
        """
@@ -3509,8 +3540,8 @@ class AsyncFTSQuery(AsyncStandardQuery):

        Typically, a single vector is passed in as the query. However, you can also
        pass in multiple vectors.  This can be useful if you want to find the nearest
-        vectors to multiple query vectors. This is not expected to be faster than
-        making multiple queries concurrently; it is just a convenience method.
+        vectors to multiple query vectors. Flat searches share one table scan across
+        the query vectors instead of issuing concurrent full scans.
        If multiple vectors are passed in then an additional column `query_index`
        will be added to the results.  This column will contain the index of the
        query vector that the result is nearest to.
@@ -3870,14 +3901,15 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        blob_paths: tuple[str, ...] = ()
        if self._table is not None:
            schema = await self._table.schema()
+            projection = _query_request_projection(req)
            blob_auto_row_id = blob_auto_row_id_for_scan(
                schema,
-                req.select,
+                projection,
                with_row_id=self._with_row_id,
            )
            if blob_auto_row_id:
                blob_paths = tuple(
-                    blob_v2_projection_sources(schema, req.select).keys()
+                    blob_v2_projection_sources(schema, projection).keys()
                )
        self._blob_auto_row_id = blob_auto_row_id
        self._blob_paths = blob_paths
@@ -36,6 +36,7 @@ from lancedb._lancedb import (
    UpdateResult,
 )
 from lancedb.embeddings.base import EmbeddingFunctionConfig
+from lancedb.expr import Expr
 from lancedb.index import (
    FTS,
    BTree,
@@ -61,6 +62,7 @@ from lancedb.table import _normalize_progress

 from ..query import (
    AnalyzePlanDistributedMetrics,
+    DocumentGranularity,
    LanceQueryBuilder,
    LanceTakeQueryBuilder,
    LanceVectorQueryBuilder,
@@ -349,6 +351,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 +374,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(
@@ -860,7 +864,7 @@ class RemoteTable(Table):

    def update(
        self,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
@@ -871,9 +875,11 @@ class RemoteTable(Table):

        Parameters
        ----------
-        where: str, optional
-            The SQL where clause to use when updating rows. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'. The filter must not be empty, or it will error.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
+            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
@@ -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
@@ -1737,7 +1744,7 @@ class Table(ABC):
    @abstractmethod
    def update(
        self,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
@@ -1752,9 +1759,11 @@ class Table(ABC):

        Parameters
        ----------
-        where: str, optional
-            The SQL where clause to use when updating rows. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'. The filter must not be empty, or it will error.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
+            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
@@ -1772,6 +1781,7 @@ class Table(ABC):
        Examples
        --------
        >>> import lancedb
+        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> data = pd.DataFrame({"x": [1, 2, 3], "vector": [[1.0, 2], [3, 4], [5, 6]]})
        >>> db = lancedb.connect("./.lancedb")
@@ -1781,7 +1791,7 @@ class Table(ABC):
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
-        >>> table.update(where="x = 2", values={"vector": [10.0, 10]})
+        >>> table.update(where=col("x") == 2, values={"vector": [10.0, 10]})
        UpdateResult(rows_updated=1, version=2)
        >>> table.to_pandas()
           x        vector
@@ -2120,12 +2130,25 @@ 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
@@ -3273,6 +3296,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.
@@ -3324,7 +3348,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(
@@ -3816,7 +3844,7 @@ class LanceTable(Table):

    def update(
        self,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
@@ -3827,9 +3855,11 @@ class LanceTable(Table):

        Parameters
        ----------
-        where: str, optional
-            The SQL where clause to use when updating rows. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'. The filter must not be empty, or it will error.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
+            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
@@ -3847,6 +3877,7 @@ class LanceTable(Table):
        Examples
        --------
        >>> import lancedb
+        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> data = pd.DataFrame({"x": [1, 2, 3], "vector": [[1.0, 2], [3, 4], [5, 6]]})
        >>> db = lancedb.connect("./.lancedb")
@@ -3856,7 +3887,7 @@ class LanceTable(Table):
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
-        >>> table.update(where="x = 2", values={"vector": [10.0, 10]})
+        >>> table.update(where=col("x") == 2, values={"vector": [10.0, 10]})
        UpdateResult(rows_updated=1, version=2)
        >>> table.to_pandas()
           x        vector
@@ -5970,7 +6001,7 @@ class AsyncTable:
        self,
        updates: Optional[Dict[str, Any]] = None,
        *,
-        where: Optional[str] = None,
+        where: Optional[Union[str, Expr]] = None,
        updates_sql: Optional[Dict[str, str]] = None,
    ) -> UpdateResult:
        """
@@ -5985,9 +6016,11 @@ class AsyncTable:
            The updates to apply.  The keys should be the name of the column to
            update.  The values should be the new values to assign.  This is
            required unless updates_sql is supplied.
-        where: str, optional
-            An SQL filter that controls which rows are updated. For example, 'x = 2'
-            or 'x IN (1, 2, 3)'.  Only rows that satisfy this filter will be udpated.
+        where: str or [Expr][lancedb.expr.Expr], optional
+            The filter condition. Can be a SQL string or a type-safe
+            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
+            [lit][lancedb.expr.lit]. Only rows that satisfy this filter will
+            be updated.
        updates_sql: dict, optional
            The updates to apply, expressed as SQL expression strings.  The keys should
            be column names. The values should be SQL expressions.  These can be SQL
@@ -6005,13 +6038,14 @@ class AsyncTable:
        --------
        >>> import asyncio
        >>> import lancedb
+        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> async def demo_update():
        ...     data = pd.DataFrame({"x": [1, 2], "vector": [[1, 2], [3, 4]]})
        ...     db = await lancedb.connect_async("./.lancedb")
        ...     table = await db.create_table("my_table", data)
        ...     # x is [1, 2], vector is [[1, 2], [3, 4]]
-        ...     await table.update({"vector": [10, 10]}, where="x = 2")
+        ...     await table.update({"vector": [10, 10]}, where=col("x") == 2)
        ...     # x is [1, 2], vector is [[1, 2], [10, 10]]
        ...     await table.update(updates_sql={"x": "x + 1"})
        ...     # x is [2, 3], vector is [[1, 2], [10, 10]]
@@ -6025,7 +6059,8 @@ class AsyncTable:
        if updates is not None:
            updates_sql = {k: value_to_sql(v) for k, v in updates.items()}

-        return await self._inner.update(updates_sql, where)
+        predicate = where.to_sql() if isinstance(where, Expr) else where
+        return await self._inner.update(updates_sql, predicate)

    async def add_columns(
        self,
@@ -105,7 +105,7 @@ def test_quickstart(tmp_path):
    tbl.create_index(num_sub_vectors=1)
    # --8<-- [end:create_index]
    # --8<-- [start:delete_rows]
-    tbl.delete('item = "fizz"')
+    tbl.delete("item = 'fizz'")
    # --8<-- [end:delete_rows]
    # --8<-- [start:drop_table]
    db.drop_table("my_table")
@@ -201,7 +201,7 @@ async def test_quickstart_async(tmp_path):
    await tbl.create_index("vector")
    # --8<-- [end:create_index_async]
    # --8<-- [start:delete_rows_async]
-    await tbl.delete('item = "fizz"')
+    await tbl.delete("item = 'fizz'")
    # --8<-- [end:delete_rows_async]
    # --8<-- [start:drop_table_async]
    await db.drop_table("my_table_async")
@@ -266,7 +266,7 @@ def test_table():
    tbl.add(pydantic_model_items)
    # --8<-- [end:add_table_from_pydantic]
    # --8<-- [start:delete_row]
-    tbl.delete('item = "fizz"')
+    tbl.delete("item = 'fizz'")
    # --8<-- [end:delete_row]
    # --8<-- [start:delete_specific_row]
    data = [
@@ -538,7 +538,7 @@ async def test_table_async():
    await async_tbl.add(pydantic_model_items)
    # --8<-- [end:add_table_async_from_pydantic]
    # --8<-- [start:delete_row_async]
-    await async_tbl.delete('item = "fizz"')
+    await async_tbl.delete("item = 'fizz'")
    # --8<-- [end:delete_row_async]
    # --8<-- [start:delete_specific_row_async]
    data = [
@@ -8,7 +8,12 @@ import pyarrow.compute as pc
 import pytest

 import lancedb
-from lancedb._blob import read_row_ids_from_hits, stash_auto_row_ids
+from lancedb._blob import (
+    blob_v2_projection_sources,
+    read_row_ids_from_hits,
+    stash_auto_row_ids,
+)
+from lancedb.expr import col
 from lancedb.index import FTS
 from lancedb.schema import blob_column_paths, blob_v2_column_paths

@@ -70,6 +75,14 @@ def test_blob_v2_column_paths_include_list_children():
    ]


+def test_blob_v2_projection_sources_use_typed_column_name():
+    schema = pa.schema([lancedb.blob("blob")])
+
+    assert blob_v2_projection_sources(schema, {"blob_alias": col("blob")}) == {
+        "blob_alias": "blob"
+    }
+
+
 def _legacy_v1_table(name):
    db = lancedb.connect("memory:///")
    schema = pa.schema(
@@ -166,6 +179,20 @@ async def test_async_table_to_pandas_descriptions_mode_omits_row_id():
    assert set(descriptor.keys()) == {"kind", "position", "size", "blob_id", "blob_uri"}


+@pytest.mark.asyncio
+async def test_async_typed_blob_projection_preserves_source_column():
+    db = await lancedb.connect_async("memory:///typed_blob_projection")
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("blob")])
+    table = await db.create_table("typed_blob_projection", schema=schema)
+    await table.add([{"id": 1, "blob": b"alpha"}])
+
+    hits = await table.query().select({"blob_alias": col("blob")}).to_arrow()
+
+    assert "_lance_row_id" in hits.schema.field("blob_alias").type.names
+    blobs = await table.fetch_blobs("blob", hits)
+    assert blobs.to_pylist() == [b"alpha"]
+
+
 def test_fetch_blobs_round_trip():
    table = _blob_table(
        "round_trip",
@@ -403,6 +430,50 @@ async def test_blob_v2_hybrid_fetch_blobs_async():
    assert {blobs[i].as_py() for i in range(len(blobs))} == {b"alpha", b"beta"}


+@pytest.mark.asyncio
+async def test_async_hybrid_typed_blob_projection_preserves_source_column():
+    db = await lancedb.connect_async("memory:///hybrid_typed_blob")
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("text", pa.utf8()),
+            pa.field("vector", pa.list_(pa.float32(), list_size=2)),
+            lancedb.blob("blob"),
+        ]
+    )
+    table = await db.create_table("hybrid_typed_blob", schema=schema)
+    await table.add(
+        [
+            {
+                "id": 1,
+                "text": "hello alpha",
+                "vector": [1.0, 0.0],
+                "blob": b"alpha",
+            },
+            {
+                "id": 2,
+                "text": "hello beta",
+                "vector": [0.9, 0.1],
+                "blob": b"beta",
+            },
+        ]
+    )
+    await table.create_index("text", config=FTS(with_position=False))
+
+    hits = await (
+        table.query()
+        .nearest_to([1.0, 0.0])
+        .nearest_to_text("hello")
+        .select({"blob_alias": col("blob")})
+        .limit(2)
+        .to_arrow()
+    )
+
+    assert "_lance_row_id" in hits.schema.field("blob_alias").type.names
+    blobs = await table.fetch_blobs("blob", hits)
+    assert {blobs[i].as_py() for i in range(len(blobs))} == {b"alpha", b"beta"}
+
+
 def test_blob_file_seek_read_and_read_range():
    payload = _identifiable_payload(1024)
    table = _blob_table("seek_read", [{"id": 1, "image": payload}])
@@ -52,7 +52,7 @@ class TestExprConstruction:
    def test_func(self):
        e = func("lower", col("name"))
        assert isinstance(e, Expr)
-        assert e.to_sql() == "lower(name)"
+        assert e.to_sql() == "lower(`name`)"

    def test_func_unknown_raises(self):
        with pytest.raises(Exception):
@@ -115,7 +115,7 @@ class TestExprOperators:
    def test_and_operator(self):
        e = (col("age") > lit(18)) & (col("status") == lit("active"))
        assert isinstance(e, Expr)
-        assert e.to_sql() == "((age > 18) AND (status = 'active'))"
+        assert e.to_sql() == "((age > 18) AND (`status` = 'active'))"

    def test_or_operator(self):
        e = (col("a") == lit(1)) | (col("b") == lit(2))
@@ -166,7 +166,7 @@ class TestExprOperators:
    def test_coerce_plain_str(self):
        e = col("name") == "alice"
        assert isinstance(e, Expr)
-        assert e.to_sql() == "(name = 'alice')"
+        assert e.to_sql() == "(`name` = 'alice')"

    def test_reflexive_comparisons(self):
        # 10 < col("age") swaps to col("age") > 10
@@ -198,85 +198,85 @@ class TestExprBytesLiteral:

    def test_bytes_equality_expr_sql(self):
        e = col("data") == lit(b"\xca\xfe")
-        assert e.to_sql() == "(data = X'CAFE')"
+        assert e.to_sql() == "(`data` = X'CAFE')"

    def test_bytes_ne_expr_sql(self):
        e = col("data") != lit(b"\xff")
-        assert e.to_sql() == "(data <> X'FF')"
+        assert e.to_sql() == "(`data` <> X'FF')"

    def test_bytes_compound_expr_sql(self):
        e = (col("data") == lit(b"\x01")) & (col("id") > lit(5))
-        assert e.to_sql() == "((data = X'01') AND (id > 5))"
+        assert e.to_sql() == "((`data` = X'01') AND (id > 5))"

    def test_bytes_in_function_call(self):
        # Regression test: binary literals inside scalar function calls
        # used to fail because DataFusion's unparser does not support Binary
        # scalars.  Now handled via a placeholder-substitution rewrite.
        e = func("contains", col("data"), lit(b"\xff"))
-        assert e.to_sql() == "contains(data, X'FF')"
+        assert e.to_sql() == "contains(`data`, X'FF')"

    def test_bytes_in_not(self):
        e = ~(col("data") == lit(b"\xff"))
-        assert e.to_sql() == "NOT (data = X'FF')"
+        assert e.to_sql() == "NOT (`data` = X'FF')"


 class TestExprStringMethods:
    def test_lower(self):
        e = col("name").lower()
        assert isinstance(e, Expr)
-        assert e.to_sql() == "lower(name)"
+        assert e.to_sql() == "lower(`name`)"

    def test_upper(self):
        e = col("name").upper()
        assert isinstance(e, Expr)
-        assert e.to_sql() == "upper(name)"
+        assert e.to_sql() == "upper(`name`)"

    def test_contains(self):
        e = col("text").contains(lit("hello"))
        assert isinstance(e, Expr)
-        assert e.to_sql() == "contains(text, 'hello')"
+        assert e.to_sql() == "contains(`text`, 'hello')"

    def test_contains_with_str_coerce(self):
        e = col("text").contains("hello")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "contains(text, 'hello')"
+        assert e.to_sql() == "contains(`text`, 'hello')"

    def test_chained_lower_eq(self):
        e = col("name").lower() == lit("alice")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "(lower(name) = 'alice')"
+        assert e.to_sql() == "(lower(`name`) = 'alice')"


 class TestExprCast:
    def test_cast_string(self):
        e = col("id").cast("string")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(id AS VARCHAR)"
+        assert e.to_sql() == "arrow_cast(id, 'Utf8')"

    def test_cast_int32(self):
        e = col("score").cast("int32")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(score AS INTEGER)"
+        assert e.to_sql() == "arrow_cast(score, 'Int32')"

    def test_cast_float64(self):
        e = col("val").cast("float64")
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(val AS DOUBLE)"
+        assert e.to_sql() == "arrow_cast(val, 'Float64')"

    def test_cast_pyarrow_type(self):
        e = col("score").cast(pa.int32())
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(score AS INTEGER)"
+        assert e.to_sql() == "arrow_cast(score, 'Int32')"

    def test_cast_pyarrow_float64(self):
        e = col("val").cast(pa.float64())
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(val AS DOUBLE)"
+        assert e.to_sql() == "arrow_cast(val, 'Float64')"

    def test_cast_pyarrow_string(self):
        e = col("id").cast(pa.string())
        assert isinstance(e, Expr)
-        assert e.to_sql() == "CAST(id AS VARCHAR)"
+        assert e.to_sql() == "arrow_cast(id, 'Utf8')"

    def test_cast_pyarrow_and_string_equivalent(self):
        # pa.int32() and "int32" should produce equivalent SQL
@@ -597,14 +597,14 @@ class TestExprIsin:
    def test_isin_strs(self):
        assert (
            col("status").isin(["active", "pending"]).to_sql()
-            == "status IN ('active', 'pending')"
+            == "`status` IN ('active', 'pending')"
        )

    def test_isin_coerces_and_mixes(self):
        assert col("id").isin([lit(1), 2]).to_sql() == "id IN (1, 2)"

    def test_isin_empty(self):
-        assert col("id").isin([]).to_sql() == "id IN ()"
+        assert col("id").isin([]).to_sql() == "false"

    def test_isin_filter(self, simple_table):
        result = simple_table.search().where(col("id").isin([1, 3, 5])).to_arrow()
@@ -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()
@@ -675,6 +675,21 @@ def test_distance_range(table: lancedb.table.Table):
    assert res["_distance"].to_pylist() == [min_dist, max_dist]


+@pytest.mark.parametrize("expression", ["1 - _distance", "1.0 - _distance"])
+def test_select_arithmetic_with_distance(table, expression):
+    result = (
+        table.search([10, 10])
+        .select({"similarity": expression, "_distance": "_distance"})
+        .distance_type("cosine")
+        .to_arrow()
+    )
+
+    assert result.schema.field("similarity").type == pa.float32()
+    assert result["similarity"].to_pylist() == pytest.approx(
+        [1 - distance for distance in result["_distance"].to_pylist()]
+    )
+
+
@pytest.mark.asyncio
 async def test_distance_range_async(table_async: AsyncTable):
    q = [0, 0]
@@ -897,6 +912,23 @@ def test_query_builder_batches(table):
    assert rs_list["id"][1] == 2


+def test_batch_vector_query_shares_filtered_flat_scan(table):
+    query = (
+        table.search([[1.0, 2.0], [3.0, 4.0]])
+        .where("id > 0", prefilter=True)
+        .limit(1)
+        .select(["id"])
+    )
+
+    plan = query.explain_plan(verbose=True)
+    assert "KNNVectorDistance: queries=2" in plan
+    assert "UnionExec" not in plan
+
+    results = query.to_arrow()
+    assert len(results) == 2
+    assert results["query_index"].to_pylist() == [0, 1]
+
+
 def test_dynamic_projection(table):
    rs = (
        LanceVectorQueryBuilder(table, [0, 0], "vector")
@@ -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"])
@@ -11,6 +11,7 @@ import warnings
 import weakref
 from concurrent.futures import ThreadPoolExecutor
 from datetime import date, datetime, timedelta
+from decimal import Decimal
 from time import sleep
 from typing import List
 from unittest.mock import patch
@@ -336,6 +337,21 @@ async def test_update_async(mem_db_async: AsyncConnection):
    assert await table.count_rows("id == 10") == 1


+@pytest.mark.asyncio
+async def test_update_expr_filter_literals_async(mem_db_async: AsyncConnection):
+    values = ["5", "4.66e-84", "it's"]
+    table = await mem_db_async.create_table(
+        "update_expr_literals",
+        data=[{"field": value, "result": "original"} for value in values],
+    )
+
+    for value in values:
+        update_res = await table.update({"result": value}, where=col("field") == value)
+        assert update_res.rows_updated == 1
+
+    assert (await table.to_arrow())["result"].to_pylist() == values
+
+
 def test_create_table(mem_db: DBConnection):
    schema = pa.schema(
        {
@@ -2343,6 +2359,148 @@ def test_update(mem_db: DBConnection):
    assert np.allclose(v, np.array([[1.2, 1.9], [1.1, 1.1]]))


+def test_update_expr_filter_literals(mem_db: DBConnection):
+    values = ["5", "4.66e-84", "it's"]
+    table = mem_db.create_table(
+        "update_expr_literals",
+        data=[{"field": value, "result": "original"} for value in values],
+    )
+
+    for value in values:
+        update_res = table.update(where=col("field") == value, values={"result": value})
+        assert update_res.rows_updated == 1
+
+    assert table.to_arrow()["result"].to_pylist() == values
+
+
+def test_update_expr_filter_preserves_typed_semantics(mem_db: DBConnection):
+    low = Decimal("1.234567890123456789")
+    high = Decimal("1.234567890123456790")
+    decimal_schema = pa.schema(
+        [("val", pa.decimal128(19, 18)), ("result", pa.string())]
+    )
+    decimal_table = mem_db.create_table(
+        "update_expr_decimal",
+        pa.table(
+            {"val": [low, high], "result": ["old", "old"]},
+            schema=decimal_schema,
+        ),
+    )
+    predicate = col("val") < lit(high)
+    assert decimal_table.search().where(predicate).to_arrow().num_rows == 1
+    result = decimal_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    keyword_table = mem_db.create_table(
+        "update_expr_keyword", [{"null": 1, "result": "old"}]
+    )
+    predicate = col("null") == 1
+    assert keyword_table.search().where(predicate).to_arrow().num_rows == 1
+    result = keyword_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    empty_in_table = mem_db.create_table(
+        "update_expr_empty_in", [{"id": 1, "result": "old"}]
+    )
+    predicate = col("id").isin([])
+    assert empty_in_table.search().where(predicate).to_arrow().num_rows == 0
+    result = empty_in_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 0
+
+    marker = "__lancedb_binary_placeholder_0__"
+    binary_schema = pa.schema(
+        [("payload", pa.binary()), ("text", pa.string()), ("result", pa.string())]
+    )
+    binary_table = mem_db.create_table(
+        "update_expr_binary",
+        pa.table(
+            {
+                "payload": [b"\x01", b"\x02"],
+                "text": ["other", marker],
+                "result": ["old", "old"],
+            },
+            schema=binary_schema,
+        ),
+    )
+    predicate = (col("payload") == lit(b"\x01")) | (col("text") == marker)
+    assert binary_table.search().where(predicate).to_arrow().num_rows == 2
+    result = binary_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 2
+
+    nonfinite_table = mem_db.create_table(
+        "update_expr_nonfinite",
+        [{"x": 1.0, "result": "old"}, {"x": 2.0, "result": "old"}],
+    )
+    predicate = col("x") < float("inf")
+    assert nonfinite_table.search().where(predicate).to_arrow().num_rows == 2
+    result = nonfinite_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 2
+
+    float16_table = mem_db.create_table(
+        "update_expr_float16",
+        [{"x": 1.0, "result": "old"}, {"x": 3.0, "result": "old"}],
+    )
+    predicate = col("x").cast(pa.float16()) < 2.0
+    assert float16_table.search().where(predicate).to_arrow().num_rows == 1
+    result = float16_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    string_cast_table = mem_db.create_table(
+        "update_expr_string_cast",
+        [{"x": 1, "result": "old"}, {"x": 2, "result": "old"}],
+    )
+    predicate = col("x").cast("string") == "1"
+    assert string_cast_table.search().where(predicate).to_arrow().num_rows == 1
+    result = string_cast_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    quoted_identifier_schema = pa.schema(
+        [("payload", pa.binary()), ("odd'name", pa.int64()), ("result", pa.string())]
+    )
+    quoted_identifier_table = mem_db.create_table(
+        "update_expr_quoted_identifier",
+        pa.table(
+            {"payload": [b"\x01"], "odd'name": [1], "result": ["old"]},
+            schema=quoted_identifier_schema,
+        ),
+    )
+    predicate = (col("payload") == lit(b"\x01")) & (col("odd'name") == 1)
+    assert quoted_identifier_table.search().where(predicate).to_arrow().num_rows == 1
+    result = quoted_identifier_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    decimal256_schema = pa.schema(
+        [("val", pa.decimal256(40, 2)), ("result", pa.string())]
+    )
+    decimal256_table = mem_db.create_table(
+        "update_expr_decimal256",
+        pa.table(
+            {
+                "val": [Decimal("1.00"), Decimal("3.00")],
+                "result": ["old", "old"],
+            },
+            schema=decimal256_schema,
+        ),
+    )
+    predicate = col("val") < lit(Decimal("2.00")).cast(pa.decimal256(40, 2))
+    assert decimal256_table.search().where(predicate).to_arrow().num_rows == 1
+    result = decimal256_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 1
+
+    binary_empty_table = mem_db.create_table(
+        "update_expr_binary_empty",
+        pa.table(
+            {"payload": [b"\x01", b"\x02"], "result": ["old", "old"]},
+            schema=pa.schema([("payload", pa.binary()), ("result", pa.string())]),
+        ),
+    )
+    predicate = (col("payload") == lit(b"\x01")).isin([])
+    assert binary_empty_table.search().where(predicate).to_arrow().num_rows == 0
+    assert predicate.to_sql() == "false"
+    result = binary_empty_table.update(where=predicate, values={"result": "new"})
+    assert result.rows_updated == 0
+
+
 def test_update_with_arrow_scalar(mem_db: DBConnection):
    schema = pa.schema({"id": pa.int64(), "vector": pa.list_(pa.float32(), 4)})
    table = mem_db.create_table("my_table", schema=schema)
@@ -130,6 +130,14 @@ impl PyExpr {

    // ── utilities ────────────────────────────────────────────────────────────

+    /// Return the referenced column name for a bare column expression.
+    fn column_name(&self) -> Option<String> {
+        match &self.0 {
+            DfExpr::Column(column) if column.relation.is_none() => Some(column.name.clone()),
+            _ => None,
+        }
+    }
+
    /// Render the expression as a SQL string (useful for debugging).
    fn to_sql(&self) -> PyResult<String> {
        lancedb::expr::expr_to_sql_string(&self.0).map_err(|e| PyValueError::new_err(e.to_string()))
@@ -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,
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;

@@ -16,8 +17,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 +77,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 +95,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 +187,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 +201,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))
@@ -292,6 +326,7 @@ pub struct PyQueryRequest {
    pub filter: Option<PyQueryFilter>,
    pub full_text_search: Option<PyLanceDB<FtsQuery>>,
    pub select: PySelect,
+    pub select_source_columns: Option<HashMap<String, String>>,
    pub fast_search: Option<bool>,
    pub with_row_id: Option<bool>,
    pub use_lsm: Option<bool>,
@@ -322,6 +357,7 @@ impl From<AnyQuery> for PyQueryRequest {
                full_text_search: query_request
                    .full_text_search
                    .map(|fts| PyLanceDB(fts.query)),
+                select_source_columns: PySelect::source_columns(&query_request.select),
                select: PySelect(query_request.select),
                fast_search: Some(query_request.fast_search),
                with_row_id: Some(query_request.with_row_id),
@@ -347,6 +383,7 @@ impl From<AnyQuery> for PyQueryRequest {
                offset: vector_query.base.offset,
                filter: vector_query.base.filter.map(PyQueryFilter),
                full_text_search: None,
+                select_source_columns: PySelect::source_columns(&vector_query.base.select),
                select: PySelect(vector_query.base.select),
                fast_search: Some(vector_query.base.fast_search),
                with_row_id: Some(vector_query.base.with_row_id),
@@ -379,6 +416,25 @@ impl From<AnyQuery> for PyQueryRequest {
 #[derive(Clone)]
 pub struct PySelect(Select);

+impl PySelect {
+    fn source_columns(select: &Select) -> Option<HashMap<String, String>> {
+        match select {
+            Select::Expr(pairs) => Some(
+                pairs
+                    .iter()
+                    .filter_map(|(output, expr)| match expr {
+                        lancedb::expr::DfExpr::Column(column) if column.relation.is_none() => {
+                            Some((output.clone(), column.name.clone()))
+                        }
+                        _ => None,
+                    })
+                    .collect(),
+            ),
+            _ => None,
+        }
+    }
+}
+
 impl<'py> IntoPyObject<'py> for PySelect {
    type Target = PyAny;
    type Output = Bound<'py, Self::Target>;
@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.38.0-beta.10"
+version = "0.38.0-beta.11"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -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();
@@ -19,6 +19,7 @@

 mod sql;

+pub(crate) use sql::canonicalize_sql_predicate;
 pub use sql::expr_to_sql_string;

 use std::sync::Arc;
@@ -156,7 +157,7 @@ mod tests {
        use datafusion_common::ScalarValue;
        let expr = col("data").eq(lit(ScalarValue::Binary(Some(vec![0xca, 0xfe]))));
        let sql = expr_to_sql_string(&expr).unwrap();
-        assert_eq!(sql, "(data = X'CAFE')");
+        assert_eq!(sql, "(`data` = X'CAFE')");
    }

    #[test]
@@ -166,7 +167,7 @@ mod tests {
        let int_expr = col("id").gt(lit(5i64));
        let combined = bin_expr.and(int_expr);
        let sql = expr_to_sql_string(&combined).unwrap();
-        assert_eq!(sql, "((data = X'01') AND (id > 5))");
+        assert_eq!(sql, "((`data` = X'01') AND (id > 5))");
    }

    #[test]
@@ -184,7 +185,7 @@ mod tests {
        // serialized correctly (regression test for placeholder rewrite path).
        let expr = contains(col("data"), lit(ScalarValue::Binary(Some(vec![0xff]))));
        let sql = expr_to_sql_string(&expr).unwrap();
-        assert_eq!(sql, "contains(data, X'FF')");
+        assert_eq!(sql, "contains(`data`, X'FF')");
    }

    #[test]
@@ -195,7 +196,7 @@ mod tests {
            .eq(lit(ScalarValue::Binary(Some(vec![0xab, 0xcd]))))
            .not();
        let sql = expr_to_sql_string(&expr).unwrap();
-        assert_eq!(sql, "NOT (data = X'ABCD')");
+        assert_eq!(sql, "NOT (`data` = X'ABCD')");
    }

    #[test]
@@ -205,6 +206,122 @@ mod tests {
        assert!(sql.contains("IN"), "expected IN in: {}", sql);
    }

+    #[test]
+    fn test_empty_is_in() {
+        let expr = is_in(col("id"), vec![]);
+        assert_eq!(expr_to_sql_string(&expr).unwrap(), "false");
+    }
+
+    #[test]
+    fn test_empty_is_in_discards_binary_children() {
+        use datafusion_common::ScalarValue;
+
+        let expr = is_in(
+            col("payload").eq(lit(ScalarValue::Binary(Some(vec![0x01])))),
+            vec![],
+        );
+        assert_eq!(expr_to_sql_string(&expr).unwrap(), "false");
+    }
+
+    #[test]
+    fn test_keyword_identifier() {
+        let expr = col("null").eq(lit(1i64));
+        assert_eq!(expr_to_sql_string(&expr).unwrap(), "(`null` = 1)");
+    }
+
+    #[test]
+    fn test_decimal_literal_preserves_type() {
+        use datafusion_common::ScalarValue;
+
+        let expr = col("val").lt(lit(ScalarValue::Decimal128(
+            Some(1_234_567_890_123_456_790),
+            19,
+            18,
+        )));
+        let sql = expr_to_sql_string(&expr).unwrap();
+        assert_eq!(
+            sql,
+            "(val < arrow_cast('1.234567890123456790', 'Decimal128(19, 18)'))"
+        );
+    }
+
+    #[test]
+    fn test_non_finite_float_literal_preserves_type() {
+        let expr = col("x").lt(lit(f64::INFINITY));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "(x < arrow_cast('inf', 'Float64'))"
+        );
+    }
+
+    #[test]
+    fn test_cast_uses_arrow_type_name() {
+        let string = expr_cast(col("x"), DataType::Utf8);
+        assert_eq!(
+            expr_to_sql_string(&string).unwrap(),
+            "arrow_cast(x, 'Utf8')"
+        );
+
+        let int32 = expr_cast(col("x"), DataType::Int32);
+        assert_eq!(
+            expr_to_sql_string(&int32).unwrap(),
+            "arrow_cast(x, 'Int32')"
+        );
+
+        let expr = expr_cast(col("x"), DataType::Float16).lt(lit(2.0));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "(arrow_cast(x, 'Float16') < 2.0)"
+        );
+
+        let decimal = expr_cast(lit("2.00"), DataType::Decimal256(40, 2));
+        assert_eq!(
+            expr_to_sql_string(&decimal).unwrap(),
+            "arrow_cast('2.00', 'Decimal256(40, 2)')"
+        );
+    }
+
+    #[test]
+    fn test_binary_placeholder_does_not_rewrite_user_string() {
+        use datafusion_common::ScalarValue;
+
+        let marker = "__lancedb_binary_placeholder_0__";
+        let expr = col("payload")
+            .eq(lit(ScalarValue::Binary(Some(vec![0x01]))))
+            .or(col("text").eq(lit(marker)));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "((payload = X'01') OR (`text` = '__lancedb_binary_placeholder_0__'))"
+        );
+    }
+
+    #[test]
+    fn test_binary_binding_skips_quoted_identifiers() {
+        use datafusion_common::ScalarValue;
+
+        let expr = col("payload")
+            .eq(lit(ScalarValue::Binary(Some(vec![0x01]))))
+            .and(col("odd'name").eq(lit(1i64)))
+            .and(col("odd`'name").eq(lit(2i64)));
+        assert_eq!(
+            expr_to_sql_string(&expr).unwrap(),
+            "(((payload = X'01') AND (`odd'name` = 1)) AND (`odd``'name` = 2))"
+        );
+    }
+
+    #[test]
+    fn test_binary_placeholder_collision_search_is_linear() {
+        use datafusion_common::ScalarValue;
+
+        let collision_shaped = format!("__lancedb_binary_placeholder_0__{}", "_".repeat(64_000));
+        let expr = col("payload")
+            .eq(lit(ScalarValue::Binary(Some(vec![0x01]))))
+            .and(col("text").eq(lit(collision_shaped.clone())));
+        let sql = expr_to_sql_string(&expr).unwrap();
+        assert!(sql.contains("X'01'"));
+        assert!(sql.contains(&format!("'{collision_shaped}'")));
+    }
+
    #[test]
    fn test_multiple_binary_literals() {
        use datafusion_common::ScalarValue;
@@ -1,10 +1,27 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors

+use std::{
+    any::TypeId,
+    collections::{HashMap, HashSet},
+};
+
+use arrow_array::types::{
+    Decimal32Type, Decimal64Type, Decimal128Type, Decimal256Type, DecimalType,
+};
+use arrow_schema::DataType;
 use datafusion_common::ScalarValue;
 use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
 use datafusion_expr::Expr;
-use datafusion_sql::unparser::{self, dialect::Dialect};
+use datafusion_functions::core::expr_fn::{
+    arrow_cast as datafusion_arrow_cast, arrow_try_cast as datafusion_arrow_try_cast,
+};
+use datafusion_sql::sqlparser::{
+    dialect::{Dialect as SqlParserDialect, GenericDialect},
+    keywords::ALL_KEYWORDS,
+    tokenizer::{Token, Tokenizer},
+};
+use datafusion_sql::unparser::{self, dialect::Dialect as UnparserDialect};

 /// Unparser dialect that matches the quoting style expected by the Lance SQL
 /// parser.  Lance uses backtick (`` ` ``) as the only delimited-identifier
@@ -19,17 +36,74 @@ use datafusion_sql::unparser::{self, dialect::Dialect};
 ///   lower-case by the SQL parser, which would break case-sensitive schemas).
 struct LanceSqlDialect;

-impl Dialect for LanceSqlDialect {
+impl UnparserDialect for LanceSqlDialect {
    fn identifier_quote_style(&self, identifier: &str) -> Option<char> {
-        let needs_quote = identifier.chars().any(|c| c.is_ascii_uppercase())
-            || !identifier
-                .chars()
-                .enumerate()
-                .all(|(i, c)| c == '_' || c.is_ascii_alphabetic() || (i > 0 && c.is_ascii_digit()));
+        let identifier_upper = identifier.to_ascii_uppercase();
+        let needs_quote =
+            (identifier_upper != "ID" && ALL_KEYWORDS.contains(&identifier_upper.as_str()))
+                || identifier.chars().any(|c| c.is_ascii_uppercase())
+                || !identifier.chars().enumerate().all(|(i, c)| {
+                    c == '_' || c.is_ascii_alphabetic() || (i > 0 && c.is_ascii_digit())
+                });
        if needs_quote { Some('`') } else { None }
    }
 }

+/// Lance's tokenizer dialect with SQL-standard double-quoted identifiers added.
+///
+/// Keep this deliberately small: Lance's parser wraps `GenericDialect` and
+/// delegates only identifier recognition, leaving every other dialect option at
+/// its default. In particular, `/*! ... */` remains an ordinary block comment.
+#[derive(Debug, Default)]
+struct PredicateDialect(GenericDialect);
+
+impl SqlParserDialect for PredicateDialect {
+    fn dialect(&self) -> TypeId {
+        self.0.dialect()
+    }
+
+    fn is_identifier_start(&self, ch: char) -> bool {
+        self.0.is_identifier_start(ch)
+    }
+
+    fn is_identifier_part(&self, ch: char) -> bool {
+        self.0.is_identifier_part(ch)
+    }
+
+    fn is_delimited_identifier_start(&self, ch: char) -> bool {
+        ch == '"' || ch == '`'
+    }
+}
+
+/// Canonicalize a raw SQL predicate for Lance's parser.
+///
+/// Lance wraps [`GenericDialect`] for identifier recognition while retaining the
+/// default dialect behavior for every other lexical option. [`PredicateDialect`]
+/// mirrors that contract and additionally recognizes `"` as an identifier
+/// delimiter, allowing this function to rewrite only those identifier tokens.
+pub fn canonicalize_sql_predicate(predicate: &str) -> crate::Result<String> {
+    let dialect = PredicateDialect::default();
+    let tokens = Tokenizer::new(&dialect, predicate)
+        .with_unescape(false)
+        .tokenize()
+        .map_err(|err| crate::Error::InvalidInput {
+            message: format!("invalid SQL predicate: {err}"),
+        })?;
+
+    Ok(tokens
+        .into_iter()
+        .map(|token| match token {
+            Token::Word(word) if word.quote_style == Some('"') => {
+                // with_unescape(false) retains doubled double quotes. Decode
+                // those before escaping any backticks for Lance's delimiter.
+                let identifier = word.value.replace("\"\"", "\"").replace('`', "``");
+                format!("`{identifier}`")
+            }
+            other => other.to_string(),
+        })
+        .collect())
+}
+
 /// Prefix for placeholder strings inserted in place of binary literals.  Chosen
 /// to be extremely unlikely to occur in user data.
 const BINARY_PLACEHOLDER_PREFIX: &str = "__lancedb_binary_placeholder_";
@@ -39,24 +113,128 @@ fn bytes_to_hex_sql(bytes: &[u8]) -> String {
    format!("X'{hex}'")
 }

-/// Returns true if *expr* contains a `Binary` or `LargeBinary` scalar literal
-/// anywhere in its subtree.  DataFusion's SQL unparser cannot serialize those
-/// variants, so we route such expressions through a placeholder-substitution
-/// path that emits SQL `X'...'` byte-string literals.
-fn has_binary_literal(expr: &Expr) -> bool {
-    let mut found = false;
+fn string_literals(expr: &Expr) -> HashSet<String> {
+    let mut literals = HashSet::new();
    let _ = expr.apply(&mut |e: &Expr| {
-        if matches!(
-            e,
-            Expr::Literal(ScalarValue::Binary(_) | ScalarValue::LargeBinary(_), _)
-        ) {
-            found = true;
-            Ok(TreeNodeRecursion::Stop)
-        } else {
-            Ok(TreeNodeRecursion::Continue)
+        if let Expr::Literal(
+            ScalarValue::Utf8(Some(value))
+            | ScalarValue::LargeUtf8(Some(value))
+            | ScalarValue::Utf8View(Some(value)),
+            _,
+        ) = e
+        {
+            literals.insert(value.clone());
        }
+        Ok(TreeNodeRecursion::Continue)
    });
-    found
+    literals
+}
+
+fn typed_string_literal(value: String, data_type: DataType) -> Expr {
+    datafusion_arrow_cast(
+        Expr::Literal(ScalarValue::Utf8(Some(value)), None),
+        Expr::Literal(ScalarValue::Utf8(Some(data_type.to_string())), None),
+    )
+}
+
+fn next_binary_placeholder(user_strings: &HashSet<String>, next_id: &mut usize) -> String {
+    loop {
+        let placeholder = format!("{BINARY_PLACEHOLDER_PREFIX}{}__", *next_id);
+        *next_id += 1;
+        if !user_strings.contains(&placeholder) {
+            return placeholder;
+        }
+    }
+}
+
+fn bind_binary_literals(
+    sql: &str,
+    mut bindings: HashMap<String, Vec<u8>>,
+) -> crate::Result<String> {
+    let bytes = sql.as_bytes();
+    let mut output = Vec::with_capacity(bytes.len());
+    let mut index = 0;
+
+    // Walk SQL string tokens once. Placeholders are plain, unescaped string
+    // literals, so this remains linear even when user strings are large or
+    // deliberately resemble the placeholder prefix.
+    while index < bytes.len() {
+        if bytes[index] == b'`' {
+            let identifier_start = index;
+            index += 1;
+            let mut identifier_end = None;
+            while index < bytes.len() {
+                if bytes[index] == b'`' {
+                    if index + 1 < bytes.len() && bytes[index + 1] == b'`' {
+                        index += 2;
+                    } else {
+                        index += 1;
+                        identifier_end = Some(index);
+                        break;
+                    }
+                } else {
+                    index += 1;
+                }
+            }
+
+            let Some(identifier_end) = identifier_end else {
+                return Err(crate::Error::InvalidInput {
+                    message: "unterminated identifier while binding binary literal".to_string(),
+                });
+            };
+            output.extend_from_slice(&bytes[identifier_start..identifier_end]);
+            continue;
+        }
+
+        if bytes[index] != b'\'' {
+            output.push(bytes[index]);
+            index += 1;
+            continue;
+        }
+
+        let literal_start = index;
+        index += 1;
+        let content_start = index;
+        let mut escaped = false;
+        let mut content_end = None;
+        while index < bytes.len() {
+            if bytes[index] == b'\'' {
+                if index + 1 < bytes.len() && bytes[index + 1] == b'\'' {
+                    escaped = true;
+                    index += 2;
+                } else {
+                    content_end = Some(index);
+                    index += 1;
+                    break;
+                }
+            } else {
+                index += 1;
+            }
+        }
+
+        let Some(content_end) = content_end else {
+            return Err(crate::Error::InvalidInput {
+                message: "unterminated string while binding binary literal".to_string(),
+            });
+        };
+
+        let placeholder = &sql[content_start..content_end];
+        if !escaped && let Some(value) = bindings.remove(placeholder) {
+            output.extend_from_slice(bytes_to_hex_sql(&value).as_bytes());
+        } else {
+            output.extend_from_slice(&bytes[literal_start..index]);
+        }
+    }
+
+    if !bindings.is_empty() {
+        return Err(crate::Error::InvalidInput {
+            message: "failed to bind binary literal while serializing expression".to_string(),
+        });
+    }
+
+    String::from_utf8(output).map_err(|e| crate::Error::InvalidInput {
+        message: format!("failed to bind binary literal: {e}"),
+    })
 }

 fn run_unparser(expr: &Expr) -> crate::Result<String> {
@@ -69,25 +247,37 @@ fn run_unparser(expr: &Expr) -> crate::Result<String> {
 }

 pub fn expr_to_sql_string(expr: &Expr) -> crate::Result<String> {
-    // Fast path: no binary literals — DataFusion's unparser handles everything.
-    if !has_binary_literal(expr) {
-        return run_unparser(expr);
-    }
-
-    // Slow path: DataFusion's unparser cannot serialize `Binary`/`LargeBinary`
-    // scalars, so we rewrite each one to a unique string-literal placeholder,
-    // let the unparser do the rest of the work, then substitute the SQL
-    // `X'...'` byte-string literal back in.  This keeps the operator/function
-    // serialization logic centralized in DataFusion and works for every
-    // expression node type the unparser supports.
-    let mut bindings: Vec<Vec<u8>> = Vec::new();
+    // DataFusion's unparser needs a few adaptations before its SQL can be
+    // reparsed by Lance without changing the typed expression's semantics:
+    //
+    // * decimal literals need an explicit cast to preserve precision and scale;
+    // * casts need exact Arrow type names rather than SQL type aliases;
+    // * an empty IN list is valid in DataFusion but invalid SQL;
+    // * binary literals are unsupported by the unparser and need placeholders.
+    // Eliminate empty membership expressions before visiting their children.
+    // Otherwise a discarded binary child could leave behind a stale binding.
    let rewritten = expr
        .clone()
+        .transform(|e: Expr| match e {
+            Expr::InList(in_list) if in_list.list.is_empty() => Ok(Transformed::yes(
+                Expr::Literal(ScalarValue::Boolean(Some(in_list.negated)), None),
+            )),
+            other => Ok(Transformed::no(other)),
+        })
+        .map_err(|e| crate::Error::InvalidInput {
+            message: format!("failed to rewrite expression: {e}"),
+        })?
+        .data;
+
+    let user_strings = string_literals(&rewritten);
+    let mut next_placeholder_id = 0;
+    let mut binary_bindings = HashMap::new();
+    let rewritten = rewritten
        .transform(|e: Expr| match e {
            Expr::Literal(ScalarValue::Binary(Some(bytes)), m)
            | Expr::Literal(ScalarValue::LargeBinary(Some(bytes)), m) => {
-                let placeholder = format!("{}{}__", BINARY_PLACEHOLDER_PREFIX, bindings.len());
-                bindings.push(bytes);
+                let placeholder = next_binary_placeholder(&user_strings, &mut next_placeholder_id);
+                binary_bindings.insert(placeholder.clone(), bytes);
                Ok(Transformed::yes(Expr::Literal(
                    ScalarValue::Utf8(Some(placeholder)),
                    m,
@@ -97,6 +287,57 @@ pub fn expr_to_sql_string(expr: &Expr) -> crate::Result<String> {
            | Expr::Literal(ScalarValue::LargeBinary(None), m) => {
                Ok(Transformed::yes(Expr::Literal(ScalarValue::Null, m)))
            }
+            Expr::Literal(ScalarValue::Decimal32(Some(value), precision, scale), _m) => {
+                let value = Decimal32Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal32(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Decimal64(Some(value), precision, scale), _m) => {
+                let value = Decimal64Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal64(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Decimal128(Some(value), precision, scale), _m) => {
+                let value = Decimal128Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal128(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Decimal256(Some(value), precision, scale), _m) => {
+                let value = Decimal256Type::format_decimal(value, precision, scale);
+                Ok(Transformed::yes(typed_string_literal(
+                    value,
+                    DataType::Decimal256(precision, scale),
+                )))
+            }
+            Expr::Literal(ScalarValue::Float16(Some(value)), _m) if !value.is_finite() => Ok(
+                Transformed::yes(typed_string_literal(value.to_string(), DataType::Float16)),
+            ),
+            Expr::Literal(ScalarValue::Float32(Some(value)), _m) if !value.is_finite() => Ok(
+                Transformed::yes(typed_string_literal(value.to_string(), DataType::Float32)),
+            ),
+            Expr::Literal(ScalarValue::Float64(Some(value)), _m) if !value.is_finite() => Ok(
+                Transformed::yes(typed_string_literal(value.to_string(), DataType::Float64)),
+            ),
+            Expr::Cast(cast) => Ok(Transformed::yes(datafusion_arrow_cast(
+                *cast.expr,
+                Expr::Literal(
+                    ScalarValue::Utf8(Some(cast.field.data_type().to_string())),
+                    None,
+                ),
+            ))),
+            Expr::TryCast(cast) => Ok(Transformed::yes(datafusion_arrow_try_cast(
+                *cast.expr,
+                Expr::Literal(
+                    ScalarValue::Utf8(Some(cast.field.data_type().to_string())),
+                    None,
+                ),
+            ))),
            other => Ok(Transformed::no(other)),
        })
        .map_err(|e| crate::Error::InvalidInput {
@@ -104,12 +345,58 @@ pub fn expr_to_sql_string(expr: &Expr) -> crate::Result<String> {
        })?
        .data;

-    let mut sql = run_unparser(&rewritten)?;
-    for (i, bytes) in bindings.iter().enumerate() {
-        // The unparser quotes string literals with single quotes, so the
-        // placeholder appears as `'__lancedb_binary_placeholder_<i>__'`.
-        let quoted = format!("'{}{}__'", BINARY_PLACEHOLDER_PREFIX, i);
-        sql = sql.replace(&quoted, &bytes_to_hex_sql(bytes));
+    let sql = run_unparser(&rewritten)?;
+    if binary_bindings.is_empty() {
+        Ok(sql)
+    } else {
+        bind_binary_literals(&sql, binary_bindings)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::canonicalize_sql_predicate;
+
+    #[test]
+    fn normalizes_double_quoted_identifiers() {
+        assert_eq!(
+            canonicalize_sql_predicate(r#""PartyAbbrev" = 'D'"#).unwrap(),
+            "`PartyAbbrev` = 'D'"
+        );
+        assert_eq!(
+            canonicalize_sql_predicate(r#""MetaData"."userId" = 5"#).unwrap(),
+            "`MetaData`.`userId` = 5"
+        );
+        assert_eq!(
+            canonicalize_sql_predicate(r#""a""b" = 1"#).unwrap(),
+            "`a\"b` = 1"
+        );
+    }
+
+    #[test]
+    fn preserves_quotes_inside_literals_and_backticks() {
+        let filter = r#"name = 'Alice "Ace"' AND `quoted"field` = 1"#;
+        assert_eq!(canonicalize_sql_predicate(filter).unwrap(), filter);
+    }
+
+    #[test]
+    fn preserves_literals_and_comments_using_lance_dialect_rules() {
+        let predicate = r#"path = '\' AND "PartyAbbrev" = 'D' -- unmatched " in comment"#;
+        assert_eq!(
+            canonicalize_sql_predicate(predicate).unwrap(),
+            r#"path = '\' AND `PartyAbbrev` = 'D' -- unmatched " in comment"#
+        );
+
+        let predicate = r#"id = 1 /* unmatched " in block comment */"#;
+        assert_eq!(canonicalize_sql_predicate(predicate).unwrap(), predicate);
+
+        let predicate = r#"id = 1 /*! OR "PartyAbbrev" = 'D' */"#;
+        assert_eq!(canonicalize_sql_predicate(predicate).unwrap(), predicate);
+    }
+
+    #[test]
+    fn rejects_unterminated_double_quoted_identifier() {
+        let error = canonicalize_sql_predicate(r#""PartyAbbrev = 'D'"#).unwrap_err();
+        assert!(matches!(error, crate::Error::InvalidInput { .. }));
    }
-    Ok(sql)
 }
@@ -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::*;
@@ -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 {
@@ -170,6 +170,15 @@ pub(crate) fn plan(
    filter: Option<&str>,
    limit: Option<u64>,
 ) -> Result<(MaterializedViewDefinition, Vec<ArrowField>, Lineage)> {
+    let filter = filter
+        .map(crate::expr::canonicalize_sql_predicate)
+        .transpose()
+        .map_err(|err| match err {
+            Error::InvalidInput { message } => Error::InvalidInput {
+                message: format!("invalid view filter: {message}"),
+            },
+            err => err,
+        })?;
    let projections: Vec<(String, String)> = if projections.is_empty() {
        source_schema
            .fields()
@@ -274,7 +283,7 @@ pub(crate) fn plan(
        declared.push(output);
    }

-    if let Some(filter) = filter {
+    if let Some(filter) = filter.as_deref() {
        let expr = planner
            .parse_filter(filter)
            .map_err(|e| Error::InvalidInput {
@@ -314,7 +323,7 @@ pub(crate) fn plan(
            .into_iter()
            .map(|(output, expression)| ViewProjection { output, expression })
            .collect(),
-        filter: filter.map(String::from),
+        filter,
        limit,
        inputs,
    };
@@ -46,8 +46,9 @@ use lance_table::format::Fragment;
 use serde::{Deserialize, Serialize};

 use super::{
-    INCARNATION_META_KEY, MaterializedViewDefinition, REFRESHED_AT_MS_META_KEY,
-    SOURCE_ROW_ID_COLUMN, SOURCE_VERSION_META_KEY,
+    DEFINITION_META_KEY, INCARNATION_META_KEY, MaterializedViewDefinition,
+    REFRESHED_AT_MS_META_KEY, SOURCE_ROW_ID_COLUMN, SOURCE_VERSION_META_KEY,
+    definition_to_metadata,
 };
 use crate::database::OpenTableRequest;
 use crate::table::{NativeTable, NativeTableExt, Table};
@@ -197,8 +198,28 @@ pub(crate) async fn execute_refresh(
            ),
        });
    }
+    let definition_changed =
+        definition.filter != replanned.filter || definition.inputs != replanned.inputs;
    let definition = &replanned;

+    // A watermark written for a legacy raw filter certifies the rows that
+    // filter produced, not the canonical predicate above. Rebuild instead of
+    // accepting or advancing it, and persist the migrated definition in the
+    // same metadata commit that certifies the replacement rows.
+    if definition_changed {
+        return rebuild(
+            view_native,
+            &view_ds,
+            &source_ds,
+            source_version,
+            source_ts,
+            definition,
+            true,
+            expected_incarnation,
+        )
+        .await;
+    }
+
    let metadata = &view_ds.schema().metadata;
    let watermark: Option<u64> = metadata
        .get(SOURCE_VERSION_META_KEY)
@@ -257,6 +278,7 @@ pub(crate) async fn execute_refresh(
                        source_version,
                        source_ts,
                        definition,
+                        false,
                        expected_incarnation,
                    )
                    .await
@@ -271,6 +293,7 @@ pub(crate) async fn execute_refresh(
                source_version,
                source_ts,
                definition,
+                false,
                expected_incarnation,
            )
            .await
@@ -683,6 +706,7 @@ async fn incremental(
            view_ds.clone(),
            source_version,
            source_ts,
+            None,
            expected_incarnation,
        )
        .await?;
@@ -704,6 +728,7 @@ async fn incremental(
            published,
            source_version,
            source_ts,
+            None,
            expected_incarnation,
        )
        .await?;
@@ -775,6 +800,7 @@ async fn incremental(
            published,
            source_version,
            source_ts,
+            None,
            expected_incarnation,
        )
        .await?;
@@ -824,12 +850,14 @@ async fn incremental(
        appended,
        source_version,
        source_ts,
+        None,
        expected_incarnation,
    )
    .await?;
    Ok(Some(result))
 }

+#[allow(clippy::too_many_arguments)]
 async fn rebuild(
    view_native: &NativeTable,
    view_ds: &Dataset,
@@ -837,6 +865,7 @@ async fn rebuild(
    source_version: u64,
    source_ts: u128,
    definition: &MaterializedViewDefinition,
+    persist_definition: bool,
    expected_incarnation: Option<&str>,
 ) -> Result<RefreshMaterializedViewResult> {
    let rows_written = Arc::new(AtomicU64::new(0));
@@ -867,6 +896,7 @@ async fn rebuild(
        replaced,
        source_version,
        source_ts,
+        persist_definition.then_some(definition),
        expected_incarnation,
    )
    .await?;
@@ -981,6 +1011,7 @@ async fn stamp_watermark(
    mut dataset: Dataset,
    source_version: u64,
    source_ts: u128,
+    definition: Option<&MaterializedViewDefinition>,
    expected_incarnation: Option<&str>,
 ) -> Result<u64> {
    ensure_incarnation(&dataset, expected_incarnation, dataset.uri()).await?;
@@ -993,27 +1024,32 @@ async fn stamp_watermark(
        .get(INCARNATION_META_KEY)
        .cloned()
        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
-    dataset
-        .update_schema_metadata([
-            (INCARNATION_META_KEY.to_string(), Some(incarnation)),
-            (
-                SOURCE_VERSION_META_KEY.to_string(),
-                Some(source_version.to_string()),
-            ),
-            (
-                SOURCE_VERSION_TS_META_KEY.to_string(),
-                Some(source_ts.to_string()),
-            ),
-            (
-                REFRESHED_AT_MS_META_KEY.to_string(),
-                Some(now_ms().to_string()),
-            ),
-            (
-                VIEW_VERSION_META_KEY.to_string(),
-                Some(predicted.to_string()),
-            ),
-        ])
-        .await?;
+    let mut metadata = vec![(INCARNATION_META_KEY.to_string(), Some(incarnation))];
+    if let Some(definition) = definition {
+        metadata.push((
+            DEFINITION_META_KEY.to_string(),
+            Some(definition_to_metadata(definition)?),
+        ));
+    }
+    metadata.extend([
+        (
+            SOURCE_VERSION_META_KEY.to_string(),
+            Some(source_version.to_string()),
+        ),
+        (
+            SOURCE_VERSION_TS_META_KEY.to_string(),
+            Some(source_ts.to_string()),
+        ),
+        (
+            REFRESHED_AT_MS_META_KEY.to_string(),
+            Some(now_ms().to_string()),
+        ),
+        (
+            VIEW_VERSION_META_KEY.to_string(),
+            Some(predicted.to_string()),
+        ),
+    ]);
+    dataset.update_schema_metadata(metadata).await?;
    let actual = dataset.version().version;
    if actual != predicted {
        return Err(Error::Runtime {
@@ -1585,6 +1621,106 @@ mod tests {
        assert_eq!(read(view.table(), "x").await, vec![20, 40]);
    }

+    #[tokio::test]
+    async fn test_mixed_case_filter_is_canonicalized_for_lineage_and_refresh() {
+        let conn = connect("memory://").execute().await.unwrap();
+        let batch = record_batch!(
+            ("id", Int32, [1, 2, 3]),
+            ("PartyAbbrev", Utf8, ["D", "R", "D"])
+        )
+        .unwrap();
+        conn.create_table("src", batch)
+            .write_options(crate::materialized_view::tests::stable_row_ids())
+            .execute()
+            .await
+            .unwrap();
+        conn.create_materialized_view("democrats", "src")
+            .select([("id", "id")])
+            .only_if(r#""PartyAbbrev" = 'D'"#)
+            .execute()
+            .await
+            .unwrap();
+
+        // Reopen from schema metadata so these assertions cover the stored
+        // predicate and lineage, not only the declaration-time handle.
+        let view = conn.open_materialized_view("democrats").await.unwrap();
+        assert_eq!(
+            view.definition().filter.as_deref(),
+            Some("`PartyAbbrev` = 'D'")
+        );
+        assert_eq!(view.definition().inputs, ["PartyAbbrev", "id"]);
+
+        let result = view.refresh().execute().await.unwrap();
+        assert_eq!(result.rows_written, 2);
+        assert_eq!(read(view.table(), "id").await, vec![1, 3]);
+    }
+
+    #[tokio::test]
+    async fn test_legacy_raw_filter_rebuilds_and_persists_canonical_definition() {
+        let conn = connect("memory://").execute().await.unwrap();
+        let batch = record_batch!(
+            ("id", Int32, [1, 2, 3]),
+            ("PartyAbbrev", Utf8, ["D", "R", "D"])
+        )
+        .unwrap();
+        conn.create_table("legacy_src", batch)
+            .write_options(crate::materialized_view::tests::stable_row_ids())
+            .execute()
+            .await
+            .unwrap();
+        let view = conn
+            .create_materialized_view("legacy_view", "legacy_src")
+            .select([("id", "id")])
+            .only_if(r#""PartyAbbrev" = 'X'"#)
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(view.refresh().execute().await.unwrap().rows_written, 0);
+
+        // Model a definition and up-to-date watermark written before filter
+        // canonicalization was applied to materialized views.
+        let mut legacy = view.definition().clone();
+        legacy.filter = Some(r#""PartyAbbrev" = 'D'"#.into());
+        legacy.inputs = vec!["id".into()];
+        let native = view.table().as_native().unwrap();
+        let mut dataset = native.dataset.get().await.unwrap().as_ref().clone();
+        let predicted = dataset.version().version + 1;
+        dataset
+            .update_schema_metadata([
+                (
+                    DEFINITION_META_KEY.to_string(),
+                    Some(definition_to_metadata(&legacy).unwrap()),
+                ),
+                (
+                    VIEW_VERSION_META_KEY.to_string(),
+                    Some(predicted.to_string()),
+                ),
+            ])
+            .await
+            .unwrap();
+        native.dataset.update(dataset);
+
+        let reopened = conn.open_materialized_view("legacy_view").await.unwrap();
+        let result = reopened.refresh().execute().await.unwrap();
+        assert_eq!(result.mode, RefreshMode::Rebuild);
+        assert_eq!(result.rows_written, 2);
+        assert_eq!(read(reopened.table(), "id").await, vec![1, 3]);
+
+        // A fresh handle proves the migration was stored alongside the new
+        // watermark and therefore happens only once.
+        let migrated = conn.open_materialized_view("legacy_view").await.unwrap();
+        assert_eq!(
+            migrated.definition().filter.as_deref(),
+            Some("`PartyAbbrev` = 'D'")
+        );
+        assert_eq!(migrated.definition().inputs, ["PartyAbbrev", "id"]);
+        assert_eq!(
+            migrated.refresh().execute().await.unwrap().mode,
+            RefreshMode::NoOp
+        );
+        assert_eq!(read(migrated.table(), "id").await, vec![1, 3]);
+    }
+
    #[tokio::test]
    async fn test_append_refreshes_incrementally() {
        let (_conn, source, view) = refreshed_doubled(vec![1, 2]).await;
@@ -2767,7 +2903,7 @@ mod tests {
        let stale = view_native.dataset.get().await.unwrap().as_ref().clone();
        view.table().delete("x = 1").await.unwrap();

-        let err = stamp_watermark(view_native, stale, 99, 99, None).await;
+        let err = stamp_watermark(view_native, stale, 99, 99, None, None).await;
        assert!(err.is_err());

        let result = view.refresh().execute().await.unwrap();
@@ -399,6 +399,9 @@ pub trait QueryBase {
    /// x > 5 OR y = 'test'
    /// ```
    ///
+    /// Identifiers may be delimited with SQL-standard double quotes or
+    /// backticks. String literals must use single quotes.
+    ///
    /// Filtering performance can often be improved by creating a scalar index
    /// on the filter column(s).
    ///
@@ -913,6 +916,17 @@ impl QueryRequest {
    /// use different representations) the error is recorded and surfaced later
    /// by [`Self::check_filter`].
    pub(crate) fn add_filter(&mut self, new: QueryFilter) {
+        let new = match new {
+            QueryFilter::Sql(filter) => match crate::expr::canonicalize_sql_predicate(&filter) {
+                Ok(filter) => QueryFilter::Sql(filter),
+                Err(err) => {
+                    self.filter_error = Some(err.to_string());
+                    return;
+                }
+            },
+            other => other,
+        };
+
        self.filter = Some(match self.filter.take() {
            None => new,
            Some(existing) => match and_filters(existing, new) {
@@ -1174,12 +1188,12 @@ impl VectorQuery {

    /// Add another query vector to the search.
    ///
-    /// Multiple searches will be dispatched as part of the query.
-    /// This is a convenience method for adding multiple query vectors
-    /// to the search. It is not expected to be faster than issuing
-    /// multiple queries concurrently.
+    /// Multiple searches will be dispatched as a batch. Flat searches share
+    /// one table scan across the query vectors, avoiding the scan and memory
+    /// amplification of issuing the searches concurrently. Indexed searches
+    /// may still perform per-vector index work.
    ///
-    /// The output data will contain an additional columns `query_index` which
+    /// The output data will contain an additional column `query_index` which
    /// will contain the index of the query vector that was used to generate the
    /// result.
    pub fn add_query_vector(mut self, vector: impl IntoQueryVector) -> Result<Self> {
@@ -1646,10 +1660,14 @@ mod tests {
    use std::{collections::HashSet, sync::Arc};

    use super::*;
-    use arrow::{array::downcast_array, compute::concat_batches, datatypes::Int32Type};
+    use arrow::{
+        array::downcast_array,
+        compute::concat_batches,
+        datatypes::{Int32Type, UInt8Type},
+    };
    use arrow_array::{
-        FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray, cast::AsArray,
-        types::Float32Type,
+        FixedSizeListArray, Float32Array, Int32Array, RecordBatch, RecordBatchIterator,
+        StringArray, cast::AsArray, types::Float32Type,
    };
    use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
    use futures::{StreamExt, TryStreamExt};
@@ -1878,6 +1896,157 @@ mod tests {
        query.execute().await.unwrap();
    }

+    #[tokio::test]
+    async fn test_double_quoted_predicates_across_table_operations() {
+        let tmp_dir = tempdir().unwrap();
+        let dataset_path = tmp_dir.path().join("test.lance");
+        let uri = dataset_path.to_str().unwrap();
+        let schema = Arc::new(ArrowSchema::new(vec![
+            ArrowField::new("id", DataType::Int32, false),
+            ArrowField::new("PartyAbbrev", DataType::Utf8, false),
+            ArrowField::new("path", DataType::Utf8, false),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
+                Arc::new(StringArray::from(vec!["D", "R", "R", "D"])),
+                Arc::new(StringArray::from(vec!["\\", "\\", "x", "x"])),
+            ],
+        )
+        .unwrap();
+
+        let conn = connect(uri).execute().await.unwrap();
+        let table = conn.create_table("parties", batch).execute().await.unwrap();
+        let batches = table
+            .query()
+            .only_if(r#""PartyAbbrev" = 'D'"#)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+
+        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+        assert_eq!(
+            table
+                .count_rows(Some(r#""PartyAbbrev" = 'D'"#.to_string()))
+                .await
+                .unwrap(),
+            2
+        );
+
+        // Public BaseTable dispatch cannot bypass canonicalization.
+        let query = AnyQuery::Query(QueryRequest {
+            filter: Some(QueryFilter::Sql(r#""PartyAbbrev" = 'D'"#.to_string())),
+            ..Default::default()
+        });
+        let batches = table
+            .base_table()
+            .query(&query, Default::default())
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+        assert_eq!(
+            table
+                .base_table()
+                .count_rows(Some(crate::table::Filter::Sql(
+                    r#""PartyAbbrev" = 'D'"#.to_string(),
+                )))
+                .await
+                .unwrap(),
+            2
+        );
+
+        for predicate in [
+            r#"id = 1 -- unmatched " in a valid SQL comment"#,
+            r#"id = 1 /* unmatched " in a valid SQL comment */"#,
+            r#"id = 1 /*! OR "PartyAbbrev" = 'D' */"#,
+            r#"path = '\' AND "PartyAbbrev" = 'D'"#,
+        ] {
+            let batches = table
+                .query()
+                .only_if(predicate)
+                .execute()
+                .await
+                .unwrap()
+                .try_collect::<Vec<_>>()
+                .await
+                .unwrap();
+            assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+        }
+
+        // The same canonical predicate contract applies to both merge filters.
+        let source = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int32Array::from(vec![1, 2, 3])),
+                Arc::new(StringArray::from(vec!["D", "R", "R"])),
+                Arc::new(StringArray::from(vec!["\\", "\\", "x"])),
+            ],
+        )
+        .unwrap();
+        let mut merge = table.merge_insert(&["id"]);
+        merge.when_not_matched_by_source_delete(Some(r#""PartyAbbrev" = 'D'"#.to_string()));
+        let result = table
+            .base_table()
+            .merge_insert(
+                merge,
+                Box::new(RecordBatchIterator::new(vec![Ok(source)], schema.clone())),
+            )
+            .await
+            .unwrap();
+        assert_eq!(result.num_deleted_rows, 1);
+
+        let source = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int32Array::from(vec![1, 2, 3])),
+                Arc::new(StringArray::from(vec!["U", "U", "U"])),
+                Arc::new(StringArray::from(vec!["\\", "\\", "x"])),
+            ],
+        )
+        .unwrap();
+        let mut merge = table.merge_insert(&["id"]);
+        merge.when_matched_update_all(Some(r#"target."PartyAbbrev" = 'D'"#.to_string()));
+        merge
+            .execute(Box::new(RecordBatchIterator::new(vec![Ok(source)], schema)))
+            .await
+            .unwrap();
+        assert_eq!(
+            table
+                .count_rows(Some(r#""PartyAbbrev" = 'U'"#.to_string()))
+                .await
+                .unwrap(),
+            1
+        );
+
+        let update = table
+            .update()
+            .only_if(r#""PartyAbbrev" = 'R'"#)
+            .column("PartyAbbrev", "'X'");
+        table.base_table().update(update).await.unwrap();
+        assert_eq!(
+            table
+                .count_rows(Some(r#""PartyAbbrev" = 'X'"#.to_string()))
+                .await
+                .unwrap(),
+            2
+        );
+
+        let result = table
+            .base_table()
+            .delete(crate::table::Predicate::String(r#""PartyAbbrev" = 'X'"#))
+            .await
+            .unwrap();
+        assert_eq!(result.num_deleted_rows, 2);
+        assert_eq!(table.count_rows(None).await.unwrap(), 1);
+    }
+
    #[tokio::test]
    async fn test_select_with_transform() {
        let batches = make_non_empty_batches();
@@ -2334,7 +2503,8 @@ mod tests {
            .limit(1);

        let plan = query.explain_plan(true).await.unwrap();
-        assert!(plan.contains("UnionExec"));
+        assert!(plan.contains("KNNVectorDistance: queries=2"));
+        assert!(!plan.contains("UnionExec"));

        let results = query
            .execute()
@@ -2349,6 +2519,100 @@ mod tests {
        // We don't guarantee order.
        assert!(query_index.values().contains(&0));
        assert!(query_index.values().contains(&1));
+
+        // Batch KNN does not support a per-query offset, so offset queries keep
+        // the legacy per-vector plan to preserve their result semantics.
+        let offset_query = table
+            .query()
+            .nearest_to(&[0.1, 0.2, 0.3, 0.4])
+            .unwrap()
+            .add_query_vector(&[0.5, 0.6, 0.7, 0.8])
+            .unwrap()
+            .limit(1)
+            .offset(1);
+        assert!(
+            offset_query
+                .explain_plan(true)
+                .await
+                .unwrap()
+                .contains("UnionExec")
+        );
+        let offset_results = offset_query
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        assert_eq!(
+            offset_results
+                .iter()
+                .map(RecordBatch::num_rows)
+                .sum::<usize>(),
+            2
+        );
+    }
+
+    #[tokio::test]
+    async fn test_multiple_binary_query_vectors() {
+        let vectors = FixedSizeListArray::from_iter_primitive::<UInt8Type, _, _>(
+            vec![
+                Some(vec![Some(0), Some(0)]),
+                Some(vec![Some(255), Some(255)]),
+            ],
+            2,
+        );
+        let schema = Arc::new(ArrowSchema::new(vec![
+            ArrowField::new("id", DataType::Int32, false),
+            ArrowField::new("vector", vectors.data_type().clone(), false),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema,
+            vec![Arc::new(Int32Array::from(vec![0, 1])), Arc::new(vectors)],
+        )
+        .unwrap();
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let table = conn
+            .create_table("binary_batch", batch)
+            .execute()
+            .await
+            .unwrap();
+        let query = table
+            .query()
+            .nearest_to(&[0.0, 0.0])
+            .unwrap()
+            .add_query_vector(&[255.0, 255.0])
+            .unwrap()
+            .distance_type(DistanceType::Hamming)
+            .limit(1);
+
+        // Binary queries retain the per-vector plan because Lance's binary
+        // nearest path requires primitive UInt8 query arrays.
+        assert!(
+            query
+                .explain_plan(true)
+                .await
+                .unwrap()
+                .contains("UnionExec")
+        );
+
+        let results = query
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let results = concat_batches(&results[0].schema(), &results).unwrap();
+        assert_eq!(results.num_rows(), 2);
+
+        let ids = results["id"].as_primitive::<Int32Type>();
+        assert!(ids.values().contains(&0));
+        assert!(ids.values().contains(&1));
+        let query_index = results["query_index"].as_primitive::<Int32Type>();
+        assert!(query_index.values().contains(&0));
+        assert!(query_index.values().contains(&1));
    }

    #[tokio::test]
@@ -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.
@@ -1139,13 +1148,26 @@ 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
    ///
    /// * `filter` if present, only count rows matching the filter
    pub async fn count_rows(&self, filter: Option<String>) -> Result<usize> {
-        self.inner.count_rows(filter.map(Filter::Sql)).await
+        let filter = filter
+            .map(|predicate| crate::expr::canonicalize_sql_predicate(&predicate).map(Filter::Sql))
+            .transpose()?;
+        self.inner.count_rows(filter).await
    }

    /// Names of the blob v2 columns in this table, in declaration order.
@@ -1345,7 +1367,13 @@ impl Table {
    /// # });
    /// ```
    pub async fn delete(&self, predicate: impl Into<Predicate<'_>>) -> Result<DeleteResult> {
-        self.inner.delete(predicate.into()).await
+        match predicate.into() {
+            Predicate::String(predicate) => {
+                let predicate = crate::expr::canonicalize_sql_predicate(predicate)?;
+                self.inner.delete(Predicate::String(&predicate)).await
+            }
+            predicate @ Predicate::Expr(_) => self.inner.delete(predicate).await,
+        }
    }

    /// Create an index on the provided column(s).
@@ -1758,7 +1786,23 @@ impl Table {
        self.inner.alter_columns(alterations).await
    }

-    /// Update per-field metadata (merges by default).
+    /// 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.
    pub async fn update_field_metadata(
        &self,
        updates: &[FieldMetadataUpdate],
@@ -3059,6 +3103,17 @@ impl BaseTable for NativeTable {
        &self.id
    }

+    async fn query_snapshot(&self) -> Result<Arc<dyn BaseTable>> {
+        let snapshot = self.dataset.new_query_snapshot().await?;
+        let mut table = self.with_dataset(snapshot);
+        // QueryTable requests do not carry a revision. A pinned snapshot must
+        // execute locally until the namespace API can accept that revision.
+        table
+            .pushdown_operations
+            .remove(&NamespaceClientPushdownOperation::QueryTable);
+        Ok(Arc::new(table))
+    }
+
    async fn version(&self) -> Result<u64> {
        Ok(self.dataset.get().await?.version().version)
    }
@@ -3193,7 +3248,10 @@ impl BaseTable for NativeTable {
        let dataset = self.dataset.get().await?;
        match filter {
            None => Ok(dataset.count_rows(None).await?),
-            Some(Filter::Sql(sql)) => Ok(dataset.count_rows(Some(sql)).await?),
+            Some(Filter::Sql(sql)) => {
+                let sql = crate::expr::canonicalize_sql_predicate(&sql)?;
+                Ok(dataset.count_rows(Some(sql)).await?)
+            }
            Some(Filter::Datafusion(_)) => Err(Error::NotSupported {
                message: "Datafusion filters are not yet supported".to_string(),
            }),
@@ -3531,7 +3589,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!(
@@ -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();
@@ -31,8 +31,9 @@ pub(crate) async fn execute_delete(
    table.dataset.ensure_mutable()?;
    match predicate {
        Predicate::String(s) => {
+            let predicate = crate::expr::canonicalize_sql_predicate(s)?;
            let mut dataset = (*table.dataset.get().await?).clone();
-            let delete_result = dataset.delete(s).boxed().await?;
+            let delete_result = dataset.delete(&predicate).boxed().await?;
            let num_deleted_rows = delete_result.num_deleted_rows;
            let version = dataset.version().version;
            table.dataset.update(dataset);
@@ -220,9 +220,32 @@ impl MergeInsertBuilder {
    ///
    /// Returns version and statistics about the merge operation including the number of rows
    /// inserted, updated, and deleted.
-    pub async fn execute(self, new_data: Box<dyn RecordBatchReader + Send>) -> Result<MergeResult> {
+    pub async fn execute(
+        mut self,
+        new_data: Box<dyn RecordBatchReader + Send>,
+    ) -> Result<MergeResult> {
+        self.canonicalize_filters()?;
        self.table.clone().merge_insert(self, new_data).await
    }
+
+    pub(crate) fn canonicalize_filters(&mut self) -> Result<()> {
+        self.when_matched_update_all_filt =
+            canonicalize_merge_filter(self.when_matched_update_all_filt.take())?;
+        self.when_not_matched_by_source_delete_filt =
+            canonicalize_merge_filter(self.when_not_matched_by_source_delete_filt.take())?;
+        Ok(())
+    }
+}
+
+fn canonicalize_merge_filter(filter: Option<MergeFilter>) -> Result<Option<MergeFilter>> {
+    filter
+        .map(|filter| match filter {
+            MergeFilter::Sql(predicate) => {
+                crate::expr::canonicalize_sql_predicate(&predicate).map(MergeFilter::Sql)
+            }
+            filter @ MergeFilter::Expr(_) => Ok(filter),
+        })
+        .transpose()
 }

 /// Internal implementation of the merge insert logic
@@ -230,9 +253,10 @@ impl MergeInsertBuilder {
 /// This logic was moved from NativeTable::merge_insert to keep table.rs clean.
 pub(crate) async fn execute_merge_insert(
    table: &NativeTable,
-    params: MergeInsertBuilder,
+    mut params: MergeInsertBuilder,
    new_data: Box<dyn RecordBatchReader + Send>,
 ) -> Result<MergeResult> {
+    params.canonicalize_filters()?;
    super::computed_columns::ensure_no_function_bindings_for_mutation(
        table.schema().await?.as_ref(),
        "merge_insert",
@@ -1056,6 +1080,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();
@@ -17,11 +17,11 @@ use arrow::array::{AsArray, FixedSizeListBuilder, Float32Builder};
 use arrow::datatypes::{Float32Type, UInt8Type};
 use arrow_array::Array;
 use arrow_schema::{DataType, Schema};
+use datafusion_common::{Column, DataFusionError, SchemaError};
 use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::repartition::RepartitionExec;
 use datafusion_physical_plan::union::UnionExec;
-use futures::future::try_join_all;
 use lance::dataset::mem_wal::DatasetMemWalExt;
 use lance::dataset::scanner::DatasetRecordBatchStream;
 use lance::dataset::scanner::Scanner;
@@ -45,6 +45,22 @@ impl AnyQuery {
            Self::VectorQuery(query) => &query.base,
        }
    }
+
+    fn base_mut(&mut self) -> &mut QueryRequest {
+        match self {
+            Self::Query(query) => query,
+            Self::VectorQuery(query) => &mut query.base,
+        }
+    }
+
+    /// Canonicalize any raw SQL filter immediately before backend dispatch.
+    pub(crate) fn canonicalized(&self) -> Result<Self> {
+        let mut query = self.clone();
+        if let Some(QueryFilter::Sql(predicate)) = &mut query.base_mut().filter {
+            *predicate = crate::expr::canonicalize_sql_predicate(predicate)?;
+        }
+        Ok(query)
+    }
 }

 //Decide between namespace or local
@@ -53,15 +69,16 @@ pub async fn execute_query(
    query: &AnyQuery,
    options: QueryExecutionOptions,
 ) -> Result<DatasetRecordBatchStream> {
+    let query = query.canonicalized()?;
    // QueryTable pushdown runs the query server-side, but only on the main
    // branch: the namespace request carries no branch yet, so a branch handle
    // must fall through to local execution.
-    if can_execute_namespace_query(table, query).await?
+    if can_execute_namespace_query(table, &query).await?
        && let Some(ref namespace_client) = table.namespace_client
    {
-        return execute_namespace_query(table, namespace_client.clone(), query, options).await;
+        return execute_namespace_query(table, namespace_client.clone(), &query, options).await;
    }
-    execute_generic_query(table, query, options).await
+    execute_generic_query(table, &query, options).await
 }

 async fn can_execute_namespace_query(table: &NativeTable, query: &AnyQuery) -> Result<bool> {
@@ -136,9 +153,10 @@ pub async fn create_plan(
    query: &AnyQuery,
    options: QueryExecutionOptions,
 ) -> Result<Arc<dyn ExecutionPlan>> {
+    let query = query.canonicalized()?;
    let query = match query {
-        AnyQuery::VectorQuery(query) => query.clone(),
-        AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query.clone()),
+        AnyQuery::VectorQuery(query) => query,
+        AnyQuery::Query(query) => VectorQueryRequest::from_plain_query(query),
    };
    query.base.check_filter()?;

@@ -170,26 +188,48 @@ pub async fn create_plan(
    let mut column = query.column.clone();

    let mut query_vector = query.query_vector.first().cloned();
+    let mut is_batch_query = false;
    if query.query_vector.len() > 1 {
        if column.is_none() {
            // Infer a vector column with the same dimension of the query vector.
-            let arrow_schema = Schema::from(ds_ref.schema());
+            let arrow_schema = Schema::from(schema);
            column = Some(default_vector_column(
                &arrow_schema,
                Some(query.query_vector[0].len() as i32),
            )?);
        }
        let vector_field = schema.field(column.as_ref().unwrap()).unwrap();
-        if let DataType::List(_) = vector_field.data_type() {
-            // Multivector handling: concatenate into FixedSizeList<FixedSizeList<_>>
+        let (_, element_type) =
+            lance::index::vector::utils::get_vector_type(schema, column.as_ref().unwrap())?;
+        let is_binary = matches!(element_type, DataType::UInt8);
+        if matches!(vector_field.data_type(), DataType::List(_))
+            || (query.base.offset.unwrap_or(0) == 0 && !is_binary)
+        {
+            // Lance distinguishes these cases from the vector column type: a
+            // list-like query against a List column is one multivector query,
+            // while the same query against a FixedSizeList column is a batch of
+            // independent queries. The batch path shares a single flat scan and
+            // bounds retained candidate data instead of running one scan per
+            // query vector.
            let vectors = query
                .query_vector
                .iter()
                .map(|arr| arr.as_ref())
                .collect::<Vec<_>>();
            let dim = vectors[0].len();
+            if let Some((query_index, actual_dim)) = vectors
+                .iter()
+                .enumerate()
+                .find_map(|(index, vector)| (vector.len() != dim).then_some((index, vector.len())))
+            {
+                return Err(Error::InvalidInput {
+                    message: format!(
+                        "query vector at index {query_index} has dimension {actual_dim}, expected {dim}"
+                    ),
+                });
+            }
            let mut fsl_builder = FixedSizeListBuilder::with_capacity(
-                Float32Builder::with_capacity(dim),
+                Float32Builder::with_capacity(dim * vectors.len()),
                dim as i32,
                vectors.len(),
            );
@@ -200,8 +240,12 @@ pub async fn create_plan(
                fsl_builder.append(true);
            }
            query_vector = Some(Arc::new(fsl_builder.finish()));
+            is_batch_query = !matches!(vector_field.data_type(), DataType::List(_));
        } else {
-            // Multiple query vectors: create a plan for each and union them
+            // Lance's batch path has no per-query offset, and its binary path
+            // requires primitive UInt8 queries rather than a fixed-size list.
+            // Keep the prior plan shape for these cases so offsets are applied
+            // per query and binary query vectors retain their primitive shape.
            let query_vecs = query.query_vector.clone();
            let plan_futures = query_vecs
                .into_iter()
@@ -214,7 +258,7 @@ pub async fn create_plan(
                    }
                })
                .collect::<Vec<_>>();
-            let plans = try_join_all(plan_futures).await?;
+            let plans = futures::future::try_join_all(plan_futures).await?;
            return create_multi_vector_plan(plans);
        }
    }
@@ -225,7 +269,7 @@ pub async fn create_plan(
        let column = if let Some(col) = column {
            col
        } else {
-            let arrow_schema = Schema::from(ds_ref.schema());
+            let arrow_schema = Schema::from(schema);
            default_vector_column(&arrow_schema, Some(query_vector.len() as i32))?
        };

@@ -251,10 +295,14 @@ pub async fn create_plan(
        }
    }

-    scanner.limit(
-        query.base.limit.map(|limit| limit as i64),
-        query.base.offset.map(|offset| offset as i64),
-    )?;
+    // For a batch query, `nearest` already applies k to each query vector.
+    // Adding Scanner's global limit would truncate the combined result to k rows.
+    if !is_batch_query {
+        scanner.limit(
+            query.base.limit.map(|limit| limit as i64),
+            query.base.offset.map(|offset| offset as i64),
+        )?;
+    }

    if let Some(ef) = query.ef {
        scanner.ef(ef);
@@ -327,7 +375,97 @@ pub async fn create_plan(
        scanner.order_by(Some(order_by.clone()))?;
    }

-    Ok(scanner.create_plan().await?)
+    scanner
+        .create_plan()
+        .await
+        .map_err(|error| enrich_lance_field_not_found(error, schema))
+}
+
+/// Replace DataFusion's top-level field candidates with qualified leaf paths.
+///
+/// DataFusion resolves nested fields but its `FieldNotFound` error only lists the
+/// top-level Arrow fields. This makes a missing leaf look unavailable even when it
+/// exists below a struct. Keep every other Lance/DataFusion error unchanged and
+/// enrich only this one schema error at the LanceDB query boundary.
+fn enrich_lance_field_not_found(
+    error: lance::Error,
+    schema: &lance_core::datatypes::Schema,
+) -> Error {
+    let Some(field) = find_missing_field(&error) else {
+        return error.into();
+    };
+    field_not_found_error(field, &Schema::from(schema))
+}
+
+fn field_not_found_diagnostic(
+    error: &(dyn std::error::Error + 'static),
+    schema: &Schema,
+) -> Option<Error> {
+    let field = find_missing_field(error)?;
+    Some(field_not_found_error(field, schema))
+}
+
+fn field_not_found_error(field: &Column, schema: &Schema) -> Error {
+    let valid_fields = leaf_field_paths(schema);
+    let mut message = format!("Schema error: No field named {}", field.quoted_flat_name());
+    if !valid_fields.is_empty() {
+        message.push_str(". Valid fields are ");
+        message.push_str(&valid_fields.join(", "));
+    }
+    message.push('.');
+
+    Error::InvalidInput { message }
+}
+
+fn find_missing_field<'a>(error: &'a (dyn std::error::Error + 'static)) -> Option<&'a Column> {
+    if let Some(DataFusionError::SchemaError(schema_error, _)) =
+        error.downcast_ref::<DataFusionError>()
+        && let SchemaError::FieldNotFound { field, .. } = schema_error.as_ref()
+    {
+        return Some(field);
+    }
+
+    error.source().and_then(find_missing_field)
+}
+
+fn leaf_field_paths(schema: &Schema) -> Vec<String> {
+    fn format_segment(segment: &str) -> String {
+        // Quote every segment instead of maintaining a SQL keyword list. Bare
+        // lowercase names such as `true` can be parsed as expressions rather
+        // than identifiers, while backticks preserve all field names in both
+        // local SQL parsers.
+        format!("`{}`", segment.replace('`', "``"))
+    }
+
+    fn visit(fields: &arrow_schema::Fields, path: &mut Vec<String>, paths: &mut Vec<String>) {
+        for field in fields {
+            // Neither local planner can address an empty field-path segment,
+            // even when it is backtick-quoted. Do not advertise leaves beneath
+            // such a segment as valid filter fields.
+            if field.name().is_empty() {
+                continue;
+            }
+            path.push(field.name().clone());
+            match field.data_type() {
+                DataType::Struct(children) if !children.is_empty() => {
+                    visit(children, path, paths);
+                }
+                _ => {
+                    paths.push(
+                        path.iter()
+                            .map(|segment| format_segment(segment))
+                            .collect::<Vec<_>>()
+                            .join("."),
+                    );
+                }
+            }
+            path.pop();
+        }
+    }
+
+    let mut paths = Vec::new();
+    visit(schema.fields(), &mut Vec::new(), &mut paths);
+    paths
 }

 //Helper functions below
@@ -687,7 +825,10 @@ async fn parse_arrow_ipc_response(bytes: bytes::Bytes) -> Result<DatasetRecordBa
 #[cfg(test)]
 #[allow(deprecated)]
 mod tests {
-    use arrow_array::{ArrayRef, FixedSizeListArray, Float32Array};
+    use arrow_array::{
+        ArrayRef, FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray,
+        StructArray,
+    };
    use futures::TryStreamExt;
    use lance_arrow::FixedSizeListArrayExt;
    use std::sync::{
@@ -696,7 +837,7 @@ mod tests {
    };

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

    fn fixed_size_list_array(values: Vec<f32>, dimension: i32) -> FixedSizeListArray {
@@ -837,7 +978,6 @@ mod tests {
    async fn test_execute_query_local_routing() {
        use crate::connect;
        use crate::table::query::execute_query;
-        use arrow_array::{Int32Array, RecordBatch};
        use arrow_schema::{DataType, Field, Schema};

        let conn = connect("memory://").execute().await.unwrap();
@@ -877,6 +1017,164 @@ mod tests {
        assert_eq!(count, 2); // 4 and 5
    }

+    #[tokio::test]
+    async fn test_missing_filter_field_lists_nested_fields_in_local_planners() {
+        use crate::connect;
+        use arrow_schema::{DataType, Field, Schema};
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let metadata = Arc::new(StructArray::from(vec![
+            (
+                Arc::new(Field::new("year", DataType::Int32, false)),
+                Arc::new(Int32Array::from(vec![2024])) as ArrayRef,
+            ),
+            (
+                Arc::new(Field::new("genre", DataType::Utf8, false)),
+                Arc::new(StringArray::from(vec!["fiction"])) as ArrayRef,
+            ),
+            (
+                Arc::new(Field::new("Title", DataType::Int32, false)),
+                Arc::new(Int32Array::from(vec![7])) as ArrayRef,
+            ),
+            (
+                Arc::new(Field::new("true", DataType::Int32, false)),
+                Arc::new(Int32Array::from(vec![8])) as ArrayRef,
+            ),
+            (
+                Arc::new(Field::new("", DataType::Int32, false)),
+                Arc::new(Int32Array::from(vec![10])) as ArrayRef,
+            ),
+        ]));
+        let vector = Arc::new(fixed_size_list_array(vec![0.0, 1.0], 2));
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("vector", vector.data_type().clone(), false),
+            Field::new("content", DataType::Utf8, false),
+            Field::new("metadata", metadata.data_type().clone(), false),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(Int32Array::from(vec![1])),
+                vector,
+                Arc::new(StringArray::from(vec!["example"])),
+                metadata,
+            ],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("nested_error", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        let error = table
+            .query()
+            .only_if("year = 2024")
+            .execute()
+            .await
+            .err()
+            .expect("query should reject the unqualified nested field");
+        let case_sensitive_path = "`metadata`.`Title`";
+        let keyword_path = "`metadata`.`true`";
+        let expected = format!(
+            "No field named year. Valid fields are `id`, `vector`, `content`, `metadata`.`year`, `metadata`.`genre`, {case_sensitive_path}, {keyword_path}."
+        );
+
+        assert!(
+            error.to_string().contains(&expected),
+            "unexpected error: {error}"
+        );
+        for (path, value) in [(case_sensitive_path, 7), (keyword_path, 8)] {
+            table
+                .query()
+                .only_if(format!("{path} = {value}"))
+                .execute()
+                .await
+                .expect("the path advertised by the diagnostic should be reusable");
+        }
+
+        table.set_unenforced_primary_key(["id"]).await.unwrap();
+        table
+            .set_lsm_write_spec(crate::table::LsmWriteSpec::unsharded())
+            .await
+            .unwrap();
+        let lsm_error = table
+            .query()
+            .only_if("year = 2024")
+            .execute()
+            .await
+            .err()
+            .expect("LSM query should reject the unqualified nested field");
+
+        assert!(
+            lsm_error.to_string().contains(&expected),
+            "unexpected LSM error: {lsm_error}"
+        );
+        for (path, value) in [(case_sensitive_path, 7), (keyword_path, 8)] {
+            table
+                .query()
+                .only_if(format!("{path} = {value}"))
+                .execute()
+                .await
+                .expect("the path advertised by the diagnostic should be reusable in LSM queries");
+        }
+    }
+
+    #[test]
+    fn test_leaf_field_paths_preserve_arbitrary_depth() {
+        use arrow_schema::{DataType, Field, Schema};
+
+        fn nested_field(path: &[&str]) -> Field {
+            let mut segments = path.iter().rev();
+            let mut field = Field::new(
+                *segments.next().expect("path must have a leaf"),
+                DataType::Int32,
+                false,
+            );
+            for segment in segments {
+                field = Field::new(*segment, DataType::Struct(vec![field].into()), false);
+            }
+            field
+        }
+
+        let schema = Schema::new(vec![
+            nested_field(&["a", "b", "c", "d", "e"]),
+            nested_field(&["metadata", "child.with.dot"]),
+            nested_field(&["metadata", "Title"]),
+            nested_field(&["metadata", "123child"]),
+            nested_field(&["metadata", "child`tick"]),
+            nested_field(&["metadata", ""]),
+            nested_field(&["", "child"]),
+        ]);
+
+        assert_eq!(
+            leaf_field_paths(&schema),
+            vec![
+                "`a`.`b`.`c`.`d`.`e`",
+                "`metadata`.`child.with.dot`",
+                "`metadata`.`Title`",
+                "`metadata`.`123child`",
+                "`metadata`.`child``tick`",
+            ]
+        );
+
+        let source = DataFusionError::SchemaError(
+            Box::new(SchemaError::FieldNotFound {
+                field: Box::new(Column::from_name("missing")),
+                valid_fields: Vec::new(),
+            }),
+            Box::new(None),
+        );
+        let error = field_not_found_diagnostic(&source, &schema).unwrap();
+        assert!(
+            error.to_string().contains(
+                "Valid fields are `a`.`b`.`c`.`d`.`e`, `metadata`.`child.with.dot`, `metadata`.`Title`, `metadata`.`123child`, `metadata`.`child``tick`"
+            ),
+            "unexpected error: {error}"
+        );
+    }
+
    #[derive(Debug, Default)]
    struct CountingNamespaceClient {
        query_table_calls: AtomicUsize,
@@ -1057,7 +1355,38 @@ mod tests {
    }

    #[tokio::test]
-    async fn test_create_plan_multivector_structure() {
+    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_batch_vector_uses_shared_scan() {
        use arrow_array::{Float32Array, RecordBatch};
        use arrow_schema::{DataType, Field, Schema};
        use datafusion_physical_plan::display::DisplayableExecutionPlan;
@@ -1084,11 +1413,18 @@ mod tests {
            .unwrap();
        let native_table = table.as_native().unwrap();

-        // This triggers the "create_multi_vector_plan" logic branch
+        // A batch of vectors against a fixed-size vector column should use
+        // Lance's native batch KNN path instead of independent scan plans.
        let q1 = Arc::new(Float32Array::from(vec![1.0, 2.0]));
        let q2 = Arc::new(Float32Array::from(vec![3.0, 4.0]));

        let req = VectorQueryRequest {
+            base: QueryRequest {
+                filter: Some(QueryFilter::Sql("id >= 0".to_string())),
+                limit: Some(1),
+                select: Select::Columns(vec!["id".to_string()]),
+                ..Default::default()
+            },
            column: Some("vector".to_string()),
            query_vector: vec![q1, q2],
            ..Default::default()
@@ -1105,19 +1441,17 @@ mod tests {
            .indent(true)
            .to_string();

-        // We expect a RepartitionExec wrapping a UnionExec
        assert!(
-            display.contains("RepartitionExec"),
-            "Plan should include Repartitioning"
+            display.contains("KNNVectorDistance: queries=2"),
+            "plan should use native batch KNN, got:\n{display}"
        );
        assert!(
-            display.contains("UnionExec"),
-            "Plan should include a Union of multiple searches"
+            !display.contains("UnionExec"),
+            "flat batch KNN should share one scan, got:\n{display}"
        );
-        // We expect the projection to add the 'query_index' column (logic inside multi_vector_plan)
        assert!(
            display.contains("query_index"),
-            "Plan should add query_index column"
+            "plan should add query_index column, got:\n{display}"
        );
    }

@@ -27,6 +27,8 @@ use std::sync::Arc;

 use arrow_array::Array;
 use arrow_schema::{DataType, Schema as ArrowSchema};
+use datafusion::common::{DataFusionError, ToDFSchema};
+use datafusion::prelude::SessionContext;
 use datafusion_physical_plan::expressions::Column;
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr};
@@ -391,7 +393,21 @@ fn base_scanner(
    }
    if let Some(filter) = &query.base.filter {
        scanner = match filter {
-            QueryFilter::Sql(sql) => scanner.filter(sql)?,
+            QueryFilter::Sql(sql) => {
+                // Parse here instead of inside `LsmScanner::filter` so the typed
+                // DataFusion `FieldNotFound` error is still available for the
+                // same nested-field enrichment used by the ordinary scanner.
+                let schema = ArrowSchema::from(dataset.schema());
+                let df_schema = schema.clone().to_dfschema().map_err(|error| {
+                    enrich_filter_error(error, &schema, "Failed to create DFSchema")
+                })?;
+                let expr = SessionContext::new()
+                    .parse_sql_expr(sql, &df_schema)
+                    .map_err(|error| {
+                        enrich_filter_error(error, &schema, "Failed to parse filter expression")
+                    })?;
+                scanner.filter_expr(expr)
+            }
            QueryFilter::Datafusion(expr) => scanner.filter_expr(expr.clone()),
            QueryFilter::Substrait(_) => {
                return Err(Error::NotSupported {
@@ -403,6 +419,12 @@ fn base_scanner(
    Ok(scanner)
 }

+fn enrich_filter_error(error: DataFusionError, schema: &ArrowSchema, context: &str) -> Error {
+    super::field_not_found_diagnostic(&error, schema).unwrap_or_else(|| Error::InvalidInput {
+        message: format!("{context}: {error}"),
+    })
+}
+
 /// Plain scan: filter / projection / limit over base ∪ SSTables ∪ in-memory.
 /// The plain scan applies limit and offset inside the planner.
 async fn plain_plan(
@@ -55,7 +55,9 @@ 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`).
+    /// Keys to set (`Some`) or delete (`None`). See
+    /// [`Table::update_field_metadata`](crate::Table::update_field_metadata) for
+    /// the conventional `lancedb:*` keys.
    pub metadata: HashMap<String, Option<String>>,
    /// If `true`, replace the field's entire metadata map instead of merging.
    pub replace: bool,
@@ -62,22 +62,33 @@ impl UpdateBuilder {
    }

    /// Executes the update operation.
-    pub async fn execute(self) -> Result<UpdateResult> {
+    pub async fn execute(mut self) -> Result<UpdateResult> {
        if self.columns.is_empty() {
            Err(Error::InvalidInput {
                message: "at least one column must be specified in an update operation".to_string(),
            })
        } else {
+            self.canonicalize_filter()?;
            self.parent.clone().update(self).await
        }
    }
+
+    pub(crate) fn canonicalize_filter(&mut self) -> Result<()> {
+        self.filter = self
+            .filter
+            .take()
+            .map(|predicate| crate::expr::canonicalize_sql_predicate(&predicate))
+            .transpose()?;
+        Ok(())
+    }
 }

 /// Internal implementation of the update logic
 pub(crate) async fn execute_update(
    table: &NativeTable,
-    update: UpdateBuilder,
+    mut update: UpdateBuilder,
 ) -> Result<UpdateResult> {
+    update.canonicalize_filter()?;
    table.dataset.ensure_mutable()?;

    // 1. Snapshot the current dataset
@@ -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
lancedb-gatefixer[bot]	9d3962686e	fix(node): accept Arrow metadata across JavaScript realms (#3904 ) ## Summary - accept genuine Arrow metadata maps created in another JavaScript realm - validate every metadata entry and clone it into a local Map - cover an Arrow 15 VM-realm table through the public fromDataToBuffer boundary - retain structural typing for nested and dictionary Arrow data ## Root cause The sanitizer used a local-realm instanceof Map check for schema and field metadata. A genuine Map created in another JavaScript realm has the required internal Map state but fails that identity check, so fromDataToBuffer rejected the foreign table before serializing its rows. ## Scope This fixes the distinct JavaScript-realm sanitizer failure identified during review. It does not establish the cause of the S3/compaction panic reported in #1525, so that issue remains open. ## Validation - pnpm test --runInBand (707 passed, 5 skipped) - pnpm test --runInBand __test__/arrow.test.ts (189 passed) - pnpm build - pnpm lint - pnpm run docs Related to #1525 <!-- lance-gatekeeper-fix:v1 agent=b522628ad3bae914eb7266ccd899d508 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-27 23:35:02 +08:00
lancedb-gatefixer[bot]	25645d82d4	feat(python): accept expressions in update filters (#3876 ) ## Summary - allow Python sync, async, and remote table updates to accept type-safe `Expr` filters - serialize expression filters before invoking the existing update implementation - cover numeric-looking text and apostrophe-containing text in sync and async regression tests ## Root cause `Table.update` was the remaining Python write path that required callers to construct a raw SQL predicate. Dynamic text interpolated without SQL literal encoding could therefore be parsed as an integer, float, or unterminated string instead of Utf8. The expression API already encodes literals safely for query and delete filters. ## Validation - `cd python && .venv/bin/pytest python/tests/test_table.py::test_update_async python/tests/test_table.py::test_update_expr_filter_literals_async python/tests/test_table.py::test_update python/tests/test_table.py::test_update_expr_filter_literals -q` - `cd python && .venv/bin/pytest python/tests/test_expr.py -q` - `cd python && .venv/bin/ruff format --check .` - `cd python && .venv/bin/ruff check .` Fixes #1869 <!-- lance-gatekeeper-fix:v1 agent=01f1e7b69c65e8b6d3b3c1e1a7918179 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-27 20:28:21 +08:00
lancedb-gatefixer[bot]	0dd9dfdfc7	test(python): cover arithmetic with distance projections (#3862 ) ## Summary - add Python regression coverage for integer and double arithmetic against the generated _distance column - merge the current main base containing Lance v11.0.0-beta.3 from #3896 - verify both expressions retain the generated scoring field Float32 type and compute the expected values ## Root cause Lance parsed dynamic projection expressions before vector search added its generated Float32 _distance field. Without a typed provisional field, expression discovery rejected mixed numeric arithmetic. Lance upstream fixed discovery and final-schema replanning in lance-format/lance#8163, and the current base consumes that fix through Lance v11.0.0-beta.3. ## Validation - uv run --extra tests pytest python/tests/test_query.py::test_select_arithmetic_with_distance -vv --maxfail=2 — 2 passed - python/.venv/bin/ruff format --check python/python/tests/test_query.py - python/.venv/bin/ruff check . Fixes #2618 <!-- lance-gatekeeper-fix:v1 agent=816a060090517471edfb73652bb5c9fe generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-27 17:10:51 +08:00
lancedb-gatefixer[bot]	d24b2dcacc	fix: show nested fields in query schema errors (#3849 ) ## Summary - enrich local query field-not-found errors with recursively qualified Arrow struct leaf paths - preserve all other Lance and DataFusion errors unchanged - add a regression test for the Python-visible filter error described in the issue ## Root cause DataFusion builds `FieldNotFound` candidates from the top-level Arrow schema even though Lance supports dotted struct-field filters. As a result, the error listed only the struct container and hid its valid nested leaves. ## Validation - `cargo test --quiet --features remote -p lancedb table::query::tests::test_missing_filter_field_lists_nested_fields -- --exact` - `cargo check --quiet --features remote --tests --examples` - `cargo clippy --quiet --features remote --tests --examples` (passes with pre-existing unrelated warnings) - `cargo fmt --all -- --check` Fixes #951 <!-- lance-gatekeeper-fix:v1 agent=7893f7a181fd8bc1ad00acc62d1a85c2 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-27 15:47:18 +08:00
lancedb-gatefixer[bot]	2deccf21cf	fix(node): read Python embedding metadata (#3836 ) ## Summary - normalize Python snake_case and TypeScript camelCase embedding metadata - use the normalized metadata for schema validation and embedding lookup - cover appending through `Table.add()` with a Python-authored schema fixture ## Root cause Python writes embedding source and vector column names as `source_column` and `vector_column`, but the TypeScript SDK only read `sourceColumn` and `vectorColumn`. The missing source name reached the add path as `undefined`, preventing JavaScript rows from being embedded and appended. ## Validation - `pnpm lint` - `pnpm test __test__/embedding.test.ts __test__/arrow.test.ts __test__/registry.test.ts --runInBand` (201 passed, 1 skipped) - `pnpm build` - `pnpm run docs` Fixes #1289 <!-- lance-gatekeeper-fix:v1 agent=b71c18a5e33d26f4d138972e91d34e66 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-27 13:42:30 +08:00
Lance Release	ead4d27bfc	Bump version: 0.38.0-beta.10 → 0.38.0-beta.11	2026-08-27 04:31:57 +00:00
lancedb-gatefixer[bot]	5153e5a023	fix(node): preserve JSON field metadata when adding data (#4064 ) ## Summary - preserve Arrow field metadata when matching record data to a provided schema - retain metadata on partially reconstructed nested struct fields - add a regression test for lance.json metadata through Arrow IPC serialization ## Root cause The TypeScript schema inferrer rebuilt fields selected from a provided schema without copying their metadata. JSON columns therefore kept their LargeBinary physical type but lost the lance.json extension marker before insert, causing the schema mismatch reported in the issue. ## Validation - pnpm lint - pnpm build - pnpm tsc - pnpm run docs - pnpm test --runInBand (18 suites and 798 tests passed; 5 tests skipped) Fixes #4062 <!-- lance-gatekeeper-fix:v1 agent=3ec52632b71563f53d199b22629f8c4f generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>	2026-08-26 16:18:57 -07:00
lancedb-gatefixer[bot]	79f626b09e	fix: support double-quoted filter identifiers (#3825 ) ## Summary - tokenize predicates with the same GenericDialect lexical rules Lance delegates to - rewrite only SQL-standard double-quoted identifier tokens to Lance backticks - apply one predicate contract to query, count, update, delete, and both merge conditions - cover mixed-case identifiers, ordinary literals, comments, and every filter-bearing table operation ## Root cause Lance plans double-quoted tokens as string literals for compatibility. As a result, `"PartyAbbrev" = 'D'` compared two literals and silently evaluated to false instead of filtering the mixed-case column. ## Validation - `cargo fmt --all -- --check` - `cargo test --locked --quiet --features remote -p lancedb expr::sql::tests` - `cargo test --locked --quiet --features remote -p lancedb test_double_quoted_predicates_across_table_operations` - `cargo check --quiet --features remote --tests --examples` - `cargo clippy --quiet --features remote --tests --examples` Fixes #2057 <!-- lance-gatekeeper-fix:v1 agent=a44b6567cfd8890abb4f7395ff71971a generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-27 06:17:28 +08:00
lancedb-gatefixer[bot]	ae81d73563	fix: share scans across batched vector queries (#3805 ) <!-- lance-gatekeeper-fix:v1 agent=d30696bc46eb32f04c9927b0792e35d3 generation=1 --> ## Summary - use the Lance native batch KNN path so fixed-size batch vector searches share one flat table scan - validate consistent query-vector dimensions and retain the per-vector plan when offsets require its existing semantics - add Rust and Python regressions and update Rust, Python, and TypeScript API documentation ## Root cause LanceDB expanded every vector in a batch into a separate scan plan and joined the plans with `UnionExec`. For unindexed tables on S3, a batch of ten vectors therefore ran ten concurrent full scans, amplifying CPU and retained data enough to produce the reported memory spike. The native Lance batch KNN path performs bounded-memory selection for all query vectors over one flat scan. LanceDB now supplies the vectors as a batch and avoids applying a global scanner limit to the combined per-query results. Batch queries with a nonzero offset keep the previous plan because the native batch API does not support per-query offsets. ## Validation - targeted Rust batch-query plan and execution tests - `cargo check --quiet --features remote --tests --examples` - `cargo clippy --quiet --features remote --tests --examples` - `cargo fmt --all -- --check` - targeted Python batch-vector regression after rebuilding the extension - Ruff formatting/checks for the touched Python files - Node.js build, lint, docs generation, and targeted batch-vector Jest test - `git diff --check` Fixes #2468 --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Co-authored-by: Xuanwo <github@xuanwo.io>	2026-08-27 04:23:28 +08:00
Dan Tasse	8b7e13b0c6	docs: add comments about metadata conventions (#4054 ) In LanceDB Enterprise, we've adopted these conventions to give some "canonical" metadata paths. This lets us display them in a certain way in the UI or let agents standardize on them, to assume they'll find info in a certain place. This PR (only comments/docs) just documents those choices.	2026-08-26 14:19:44 -04:00
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