Merge main into gatekeeper/fix-3530-1

## 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>

.bumpversion.toml

@@ -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*)\\.

Cargo.lock

@@ -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",

docs/src/java/java.md

@@ -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>
 ```
 

java/lancedb-core/pom.xml

@@ -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>
 

java/pom.xml

@@ -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>

nodejs/Cargo.toml

@@ -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

nodejs/__test__/arrow.test.ts

@@ -6,6 +6,9 @@ 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 +39,24 @@ function sampleRecords(): Array<Record<string, any>> {
     },
   ];
 }
+
+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",
   (

nodejs/__test__/embedding.test.ts

@@ -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")

nodejs/__test__/table.test.ts

@@ -3561,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: [

nodejs/lancedb/schema.ts

@@ -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;

nodejs/npm/darwin-arm64/package.json

@@ -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",

nodejs/npm/linux-arm64-gnu/package.json

@@ -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",

nodejs/npm/linux-arm64-musl/package.json

@@ -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",

nodejs/npm/linux-x64-gnu/package.json

@@ -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",

nodejs/npm/linux-x64-musl/package.json

@@ -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",

nodejs/npm/win32-arm64-msvc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.38.0-beta.10",
+  "version": "0.38.0-beta.11",
   "os": [
     "win32"
   ],

nodejs/npm/win32-x64-msvc/package.json

@@ -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",

nodejs/package-lock.json

@@ -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"

nodejs/package.json

@@ -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",

python/Cargo.toml

@@ -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"

rust/lancedb/Cargo.toml

@@ -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

rust/lancedb/src/table/merge.rs

@@ -1431,4 +1431,195 @@ mod lsm_tests {
             "LSM vector search must rank the memtable row first"
         );
     }
+
+    #[tokio::test]
+    async fn lsm_cosine_distance_scale_and_mixed_tier_ordering() {
+        use arrow::array::{FixedSizeListBuilder, Float32Builder};
+        use arrow::datatypes::Float32Type;
+
+        use crate::index::Index;
+        use crate::index::vector::IvfPqIndexBuilder;
+
+        const DIM: usize = 8;
+        const N: usize = 256;
+
+        fn normalized_vector(state: &mut u64) -> Vec<f32> {
+            let mut vector = (0..DIM)
+                .map(|_| {
+                    *state = state
+                        .wrapping_mul(6_364_136_223_846_793_005)
+                        .wrapping_add(1);
+                    ((*state >> 32) as u32 as f32 / u32::MAX as f32) * 2.0 - 1.0
+                })
+                .collect::<Vec<_>>();
+            let norm = vector.iter().map(|value| value * value).sum::<f32>().sqrt();
+            vector.iter_mut().for_each(|value| *value /= norm);
+            vector
+        }
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new(
+                "vec",
+                DataType::FixedSizeList(
+                    Arc::new(Field::new("item", DataType::Float32, true)),
+                    DIM as i32,
+                ),
+                false,
+            ),
+        ]));
+        let make_batch = |rows: Vec<(i64, Vec<f32>)>| {
+            let ids = rows.iter().map(|(id, _)| *id).collect::<Vec<_>>();
+            let mut vectors = FixedSizeListBuilder::new(Float32Builder::new(), DIM as i32);
+            for (_, vector) in &rows {
+                vectors.values().append_slice(vector);
+                vectors.append(true);
+            }
+            RecordBatch::try_new(
+                schema.clone(),
+                vec![Arc::new(Int64Array::from(ids)), Arc::new(vectors.finish())],
+            )
+            .unwrap()
+        };
+        let first_result = |batches: &[RecordBatch]| {
+            let batch = &batches[0];
+            let id = batch["id"].as_primitive::<Int64Type>().value(0);
+            let distance = batch["_distance"].as_primitive::<Float32Type>().value(0);
+            (id, distance)
+        };
+
+        let mut state = 42;
+        let base_rows = (0..N)
+            .map(|id| (id as i64, normalized_vector(&mut state)))
+            .collect::<Vec<_>>();
+        let query = normalized_vector(&mut state);
+
+        let dir = tempdir().unwrap();
+        let conn = connect(dir.path().to_str().unwrap())
+            .execute()
+            .await
+            .unwrap();
+        let base = make_batch(base_rows);
+        let reader: Box<dyn RecordBatchReader + Send> =
+            Box::new(RecordBatchIterator::new(vec![Ok(base)], schema.clone()));
+        let table = conn
+            .create_table("cosine_lsm", reader)
+            .execute()
+            .await
+            .unwrap();
+        table.set_unenforced_primary_key(["id"]).await.unwrap();
+        table
+            .create_index(
+                &["vec"],
+                Index::IvfPq(
+                    IvfPqIndexBuilder::default()
+                        .distance_type(crate::DistanceType::Cosine)
+                        .num_partitions(1)
+                        .num_sub_vectors(1),
+                ),
+            )
+            .name("vec_cosine".to_string())
+            .execute()
+            .await
+            .unwrap();
+        table
+            .set_lsm_write_spec(
+                LsmWriteSpec::unsharded().with_maintained_indexes(vec!["vec_cosine".to_string()]),
+            )
+            .await
+            .unwrap();
+
+        let base_only = table
+            .query()
+            .nearest_to(query.as_slice())
+            .unwrap()
+            .limit(1)
+            .use_lsm(false)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let (base_id, public_distance) = first_result(&base_only);
+
+        let lsm = table
+            .query()
+            .nearest_to(query.as_slice())
+            .unwrap()
+            .limit(1)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let (lsm_id, lsm_distance) = first_result(&lsm);
+        assert_eq!(lsm_id, base_id);
+        assert!(
+            (lsm_distance - public_distance).abs() < 1e-5,
+            "LSM cosine distance {lsm_distance} did not use the public scale {public_distance}"
+        );
+
+        // Add an exact memtable result whose distance lies between the public ANN
+        // score and its doubled internal score. Correctly normalized plans still
+        // rank the ANN row first; mixed units would incorrectly rank this row first.
+        assert!(public_distance > 0.0 && public_distance < 4.0 / 3.0);
+        let memtable_distance = public_distance * 1.5;
+        let cosine_similarity = 1.0 - memtable_distance;
+        let mut orthogonal = normalized_vector(&mut state);
+        let projection = orthogonal
+            .iter()
+            .zip(&query)
+            .map(|(left, right)| left * right)
+            .sum::<f32>();
+        for (value, query_value) in orthogonal.iter_mut().zip(&query) {
+            *value -= projection * query_value;
+        }
+        let norm = orthogonal
+            .iter()
+            .map(|value| value * value)
+            .sum::<f32>()
+            .sqrt();
+        orthogonal.iter_mut().for_each(|value| *value /= norm);
+        let sine = (1.0 - cosine_similarity * cosine_similarity).sqrt();
+        let memtable_vector = query
+            .iter()
+            .zip(&orthogonal)
+            .map(|(query_value, orthogonal_value)| {
+                cosine_similarity * query_value + sine * orthogonal_value
+            })
+            .collect::<Vec<_>>();
+
+        let mut merge = table.merge_insert(&[]);
+        merge
+            .when_matched_update_all(None)
+            .when_not_matched_insert_all();
+        let memtable = make_batch(vec![(N as i64, memtable_vector)]);
+        merge
+            .execute(Box::new(RecordBatchIterator::new(
+                vec![Ok(memtable)],
+                schema,
+            )))
+            .await
+            .unwrap();
+
+        let mixed = table
+            .query()
+            .nearest_to(query.as_slice())
+            .unwrap()
+            .limit(1)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let (mixed_id, mixed_distance) = first_result(&mixed);
+        assert_eq!(
+            mixed_id, base_id,
+            "mixed LSM tiers must compare ANN and exact distances in public units"
+        );
+        assert!((mixed_distance - public_distance).abs() < 1e-5);
+    }
 }

rust/lancedb/src/table/query.rs

@@ -1,7 +1,10 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-use std::sync::Arc;
+use std::{
+    collections::{HashSet, VecDeque},
+    sync::Arc,
+};
 
 mod lsm;
 
@@ -17,14 +20,23 @@ use arrow::array::{AsArray, FixedSizeListBuilder, Float32Builder};
 use arrow::datatypes::{Float32Type, UInt8Type};
 use arrow_array::Array;
 use arrow_schema::{DataType, Schema};
-use datafusion_physical_plan::ExecutionPlan;
+use datafusion_common::{Column, DataFusionError, ScalarValue, SchemaError};
+use datafusion_expr::Operator;
+use datafusion_physical_expr::expressions::{BinaryExpr, Column as PhysicalColumn, Literal};
+use datafusion_physical_plan::PhysicalExpr;
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::repartition::RepartitionExec;
 use datafusion_physical_plan::union::UnionExec;
+use datafusion_physical_plan::{ExecutionPlan, with_new_children_if_necessary};
 use lance::dataset::mem_wal::DatasetMemWalExt;
 use lance::dataset::scanner::DatasetRecordBatchStream;
 use lance::dataset::scanner::Scanner;
+use lance::index::DatasetIndexInternalExt;
+use lance::io::exec::ANNIvfSubIndexExec;
 use lance_datafusion::exec::{analyze_plan as lance_analyze_plan, execute_plan};
+use lance_index::metrics::NoOpMetricsCollector;
+use lance_index::vector::{DIST_COL, quantizer::QuantizationType};
+use lance_linalg::distance::DistanceType as LanceDistanceType;
 use lance_namespace::LanceNamespace;
 use lance_namespace::models::{
     QueryTableRequest as NsQueryTableRequest, QueryTableRequestColumns,
@@ -191,7 +203,7 @@ pub async fn create_plan(
     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),
@@ -268,7 +280,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))?
         };
 
@@ -374,11 +386,299 @@ pub async fn create_plan(
         scanner.order_by(Some(order_by.clone()))?;
     }
 
-    Ok(scanner.create_plan().await?)
+    let mut plan = scanner
+        .create_plan()
+        .await
+        .map_err(|error| enrich_lance_field_not_found(error, schema))?;
+    let normalized_l2_indices = normalized_l2_ann_indices(plan.as_ref()).await?;
+    if !normalized_l2_indices.is_empty() {
+        // Rebuild only the affected ANN nodes with internal normalized squared-L2
+        // bounds. Exact branches keep the public cosine bounds from `plan`.
+        let internal_plan = if query.lower_bound.is_some() || query.upper_bound.is_some() {
+            scanner.distance_range(
+                query.lower_bound.map(|bound| bound / COSINE_ANN_SCALE),
+                query.upper_bound.map(|bound| bound / COSINE_ANN_SCALE),
+            );
+            scanner
+                .create_plan()
+                .await
+                .map_err(|error| enrich_lance_field_not_found(error, schema))?
+        } else {
+            plan.clone()
+        };
+        plan = normalize_ann_branches(plan, internal_plan, &normalized_l2_indices)?;
+    }
+
+    Ok(plan)
+}
+
+/// 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
 
+const COSINE_ANN_SCALE: f32 = 0.5;
+
+/// Find ANN index segments whose scores use normalized squared L2 for cosine search.
+///
+/// Cosine PQ/SQ/RQ indices normalize their vectors and use squared L2 internally.  This
+/// preserves ranking, but squared L2 over unit vectors is twice the cosine distance.  Flat
+/// cosine indices calculate cosine directly, so they are not included.
+async fn normalized_l2_ann_indices(plan: &dyn ExecutionPlan) -> Result<HashSet<String>> {
+    let mut ann_plans = Vec::new();
+    find_ann_plans(plan, &mut ann_plans);
+
+    let mut checked = HashSet::new();
+    let mut normalized_l2 = HashSet::new();
+    for ann in ann_plans {
+        if ann.query().metric_type != Some(LanceDistanceType::Cosine) {
+            continue;
+        }
+        for index in ann.indices() {
+            let uuid = index.uuid.to_string();
+            if !checked.insert(uuid.clone()) {
+                continue;
+            }
+            let vector_index = ann
+                .dataset()
+                .open_vector_index(&ann.query().column, &index.uuid, &NoOpMetricsCollector)
+                .await?;
+            let (_, quantization_type) = vector_index.sub_index_type();
+            if matches!(
+                quantization_type,
+                QuantizationType::Product | QuantizationType::Scalar | QuantizationType::Rabit
+            ) {
+                normalized_l2.insert(uuid);
+            }
+        }
+    }
+    Ok(normalized_l2)
+}
+
+/// Normalize affected ANN outputs before their parent plan nodes consume them.
+///
+/// This is used by planners that do not support distance ranges, such as the MemWAL
+/// LSM planner. The standard scanner path rebuilds a second plan when it also needs
+/// to translate range bounds, then calls [`normalize_ann_branches`] directly.
+pub(super) async fn normalize_cosine_ann_branches(
+    plan: Arc<dyn ExecutionPlan>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    let normalized_l2_indices = normalized_l2_ann_indices(plan.as_ref()).await?;
+    if normalized_l2_indices.is_empty() {
+        return Ok(plan);
+    }
+    normalize_ann_branches(plan.clone(), plan, &normalized_l2_indices)
+}
+
+fn find_ann_plans<'a>(plan: &'a dyn ExecutionPlan, ann_plans: &mut Vec<&'a ANNIvfSubIndexExec>) {
+    if let Some(ann) = plan.downcast_ref::<ANNIvfSubIndexExec>() {
+        ann_plans.push(ann);
+    }
+    for child in plan.children() {
+        find_ann_plans(child.as_ref(), ann_plans);
+    }
+}
+
+fn collect_ann_plans(
+    plan: &Arc<dyn ExecutionPlan>,
+    ann_plans: &mut VecDeque<Arc<dyn ExecutionPlan>>,
+) {
+    if plan.downcast_ref::<ANNIvfSubIndexExec>().is_some() {
+        ann_plans.push_back(plan.clone());
+        return;
+    }
+    for child in plan.children() {
+        collect_ann_plans(child, ann_plans);
+    }
+}
+
+/// Replace normalized-L2 ANN nodes with equivalent nodes that use internal bounds, then
+/// convert their output to the public cosine scale before any generic plan node consumes it.
+fn normalize_ann_branches(
+    public_plan: Arc<dyn ExecutionPlan>,
+    internal_plan: Arc<dyn ExecutionPlan>,
+    normalized_l2_indices: &HashSet<String>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    let mut internal_ann_plans = VecDeque::new();
+    collect_ann_plans(&internal_plan, &mut internal_ann_plans);
+    let normalized =
+        replace_ann_branches(public_plan, &mut internal_ann_plans, normalized_l2_indices)?;
+    if !internal_ann_plans.is_empty() {
+        return Err(Error::Runtime {
+            message: "internal and public vector plans contained different ANN branches"
+                .to_string(),
+        });
+    }
+    Ok(normalized)
+}
+
+fn replace_ann_branches(
+    public_plan: Arc<dyn ExecutionPlan>,
+    internal_ann_plans: &mut VecDeque<Arc<dyn ExecutionPlan>>,
+    normalized_l2_indices: &HashSet<String>,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    if let Some(public_ann) = public_plan.downcast_ref::<ANNIvfSubIndexExec>() {
+        let internal_plan = internal_ann_plans
+            .pop_front()
+            .ok_or_else(|| Error::Runtime {
+                message: "internal vector plan was missing an ANN branch".to_string(),
+            })?;
+        let internal_ann = internal_plan
+            .downcast_ref::<ANNIvfSubIndexExec>()
+            .expect("collected only ANN plans");
+        let same_indices = public_ann
+            .indices()
+            .iter()
+            .map(|index| &index.uuid)
+            .eq(internal_ann.indices().iter().map(|index| &index.uuid));
+        if public_ann.query().column != internal_ann.query().column
+            || public_ann.query().metric_type != internal_ann.query().metric_type
+            || !same_indices
+        {
+            return Err(Error::Runtime {
+                message: "internal and public vector plans had mismatched ANN branches".to_string(),
+            });
+        }
+
+        let normalized_count = public_ann
+            .indices()
+            .iter()
+            .filter(|index| normalized_l2_indices.contains(&index.uuid.to_string()))
+            .count();
+        if normalized_count == 0 {
+            return Ok(public_plan);
+        }
+        if normalized_count != public_ann.indices().len() {
+            return Err(Error::Runtime {
+                message: "one ANN branch mixed public and normalized-L2 distance scales"
+                    .to_string(),
+            });
+        }
+        return scale_distance_column(internal_plan, COSINE_ANN_SCALE);
+    }
+
+    let children = public_plan
+        .children()
+        .into_iter()
+        .cloned()
+        .map(|child| replace_ann_branches(child, internal_ann_plans, normalized_l2_indices))
+        .collect::<Result<Vec<_>>>()?;
+    Ok(with_new_children_if_necessary(public_plan, children)?)
+}
+
+fn scale_distance_column(
+    plan: Arc<dyn ExecutionPlan>,
+    scale: f32,
+) -> Result<Arc<dyn ExecutionPlan>> {
+    let schema = plan.schema();
+    if schema.column_with_name(DIST_COL).is_none() {
+        return Ok(plan);
+    }
+
+    let expressions: Vec<(Arc<dyn PhysicalExpr>, String)> = schema
+        .fields()
+        .iter()
+        .enumerate()
+        .map(|(index, field)| {
+            let column: Arc<dyn PhysicalExpr> = Arc::new(PhysicalColumn::new(field.name(), index));
+            let expression = if field.name() == DIST_COL {
+                let scale: Arc<dyn PhysicalExpr> =
+                    Arc::new(Literal::new(ScalarValue::Float32(Some(scale))));
+                Arc::new(BinaryExpr::new(column, Operator::Multiply, scale))
+                    as Arc<dyn PhysicalExpr>
+            } else {
+                column
+            };
+            (expression, field.name().clone())
+        })
+        .collect();
+    Ok(Arc::new(ProjectionExec::try_new(expressions, plan)?))
+}
+
 // Take many execution plans and map them into a single plan that adds
 // a query_index column and unions them.
 pub(crate) fn create_multi_vector_plan(
@@ -734,7 +1034,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::{
@@ -743,7 +1046,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 {
@@ -884,7 +1187,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();
@@ -924,6 +1226,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,
@@ -1204,6 +1664,206 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_cosine_pq_distance_uses_public_cosine_scale() {
+        use arrow_array::{Int32Array, RecordBatch, types::Float32Type};
+        use arrow_schema::{DataType, Field, Schema};
+
+        use crate::connect;
+        use crate::index::{Index, vector::IvfPqIndexBuilder};
+
+        fn normalized_vector(state: &mut u64, dimension: usize) -> Vec<f32> {
+            let mut vector = (0..dimension)
+                .map(|_| {
+                    *state = state
+                        .wrapping_mul(6_364_136_223_846_793_005)
+                        .wrapping_add(1);
+                    ((*state >> 32) as u32 as f32 / u32::MAX as f32) * 2.0 - 1.0
+                })
+                .collect::<Vec<_>>();
+            let norm = vector.iter().map(|value| value * value).sum::<f32>().sqrt();
+            vector.iter_mut().for_each(|value| *value /= norm);
+            vector
+        }
+
+        fn distances(batches: &[RecordBatch]) -> Vec<f32> {
+            batches
+                .iter()
+                .flat_map(|batch| {
+                    batch[DIST_COL]
+                        .as_primitive::<Float32Type>()
+                        .values()
+                        .to_vec()
+                })
+                .collect()
+        }
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let dimension = 8;
+        let num_rows = 256;
+        let mut state = 42;
+        let values = (0..num_rows)
+            .flat_map(|_| normalized_vector(&mut state, dimension))
+            .collect::<Vec<_>>();
+        let query_vector = normalized_vector(&mut state, dimension);
+        let vectors = Arc::new(fixed_size_list_array(values, dimension as i32));
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("vector", vectors.data_type().clone(), false),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![Arc::new(Int32Array::from_iter_values(0..num_rows)), vectors],
+        )
+        .unwrap();
+        let table = conn
+            .create_table("test_cosine_pq_distance", batch)
+            .execute()
+            .await
+            .unwrap();
+        table
+            .create_index(
+                &["vector"],
+                Index::IvfPq(
+                    IvfPqIndexBuilder::default()
+                        .distance_type(crate::DistanceType::Cosine)
+                        .num_partitions(1)
+                        .num_sub_vectors(1),
+                ),
+            )
+            .execute()
+            .await
+            .unwrap();
+
+        let approximate = table
+            .vector_search(query_vector.as_slice())
+            .unwrap()
+            .limit(5)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let refined = table
+            .vector_search(query_vector.as_slice())
+            .unwrap()
+            .limit(5)
+            .refine_factor(1)
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let approximate_distances = distances(&approximate);
+        let refined_distances = distances(&refined);
+        assert_eq!(approximate_distances.len(), refined_distances.len());
+        for (approximate, refined) in approximate_distances.iter().zip(&refined_distances) {
+            assert!(
+                (approximate - refined).abs() < 1e-5,
+                "approximate cosine distance {approximate} did not use the public scale; refined distance was {refined}"
+            );
+        }
+
+        // Distance range bounds are public cosine distances too. Lance applies them to
+        // internal ANN scores, so the planner must translate the bounds before execution.
+        let nearest = approximate_distances[0];
+        let ranged = table
+            .vector_search(query_vector.as_slice())
+            .unwrap()
+            .limit(1)
+            .distance_range(Some(nearest - 1e-5), Some(nearest + 1e-5))
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let ranged_distances = distances(&ranged);
+        assert_eq!(ranged_distances.len(), 1);
+        assert!((ranged_distances[0] - nearest).abs() < 1e-5);
+
+        let refined_ranged = table
+            .vector_search(query_vector.as_slice())
+            .unwrap()
+            .limit(1)
+            .refine_factor(1)
+            .distance_range(None, Some(nearest + 1e-5))
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        assert_eq!(
+            distances(&refined_ranged).len(),
+            1,
+            "refinement must not apply public cosine bounds to internal ANN scores"
+        );
+
+        let aliased = table
+            .vector_search(query_vector.as_slice())
+            .unwrap()
+            .limit(1)
+            .select(Select::dynamic(&[("aliased_distance", "_distance")]))
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let batch = &aliased[0];
+        let aliased_distance = batch["aliased_distance"]
+            .as_primitive::<Float32Type>()
+            .value(0);
+        let public_distance = batch[DIST_COL].as_primitive::<Float32Type>().value(0);
+        assert!(
+            (aliased_distance - public_distance).abs() < 1e-5,
+            "distance aliases and auto-projected distances must use the same public scale"
+        );
+
+        // Appended rows take an exact fallback branch. Its public range filter must stay
+        // independent of the translated ANN bounds before both branches are merged.
+        let mut orthogonal = normalized_vector(&mut state, dimension);
+        let projection = orthogonal
+            .iter()
+            .zip(&query_vector)
+            .map(|(left, right)| left * right)
+            .sum::<f32>();
+        for (value, query_value) in orthogonal.iter_mut().zip(&query_vector) {
+            *value -= projection * query_value;
+        }
+        let norm = orthogonal
+            .iter()
+            .map(|value| value * value)
+            .sum::<f32>()
+            .sqrt();
+        orthogonal.iter_mut().for_each(|value| *value /= norm);
+        let appended_vectors = Arc::new(fixed_size_list_array(orthogonal, dimension as i32));
+        let appended = RecordBatch::try_new(
+            schema,
+            vec![Arc::new(Int32Array::from(vec![num_rows])), appended_vectors],
+        )
+        .unwrap();
+        table.add(appended).execute().await.unwrap();
+
+        let mixed = table
+            .vector_search(query_vector.as_slice())
+            .unwrap()
+            .limit(5)
+            .distance_range(None, Some(nearest + 1e-5))
+            .execute()
+            .await
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        let mixed_distances = distances(&mixed);
+        assert_eq!(mixed_distances.len(), 1);
+        assert!((mixed_distances[0] - nearest).abs() < 1e-5);
+    }
+
     #[tokio::test]
     async fn test_create_plan_applies_approx_mode_to_ann_query() {
         use arrow_array::RecordBatch;

rust/lancedb/src/table/query/lsm.rs

@@ -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};
@@ -128,6 +130,10 @@ pub(super) async fn create_lsm_plan(
         .await?
     };
 
+    // Normalize cosine ANN arms before LSM merge and sort nodes compare their
+    // distances with exact SSTable and memtable arms.
+    let plan = super::normalize_cosine_ann_branches(plan).await?;
+
     // Lance appends the primary-key columns internally for dedup and keeps them in
     // the output; drop the ones the user did not request so the projection matches.
     restore_projection(plan, &query, &pk_columns)
@@ -391,7 +397,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 +423,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(