fix: preserve case in generated field path expressions (#7698)

Fixes #7697

## Summary

Fix Lance-generated DataFusion field-path expressions so schema-derived
column names preserve exact casing.

Previously, `field_path_to_expr` used DataFusion `col(...)`, which
lowercases unquoted identifiers. This caused generated nullable-vector
filters like `VECTOR IS NOT NULL` to resolve as `vector`, breaking
case-sensitive schemas during
  vector index creation / append optimization.

## Changes

- Build root field-path expressions with
`Expr::Column(Column::new_unqualified(...))` instead of `col(...)`.
- Preserve existing nested field handling through `field_newstyle(...)`.
  - Add regression coverage for:
    - uppercase nullable vector column indexing / append optimization
    - exact-case root column expression generation
    - mixed-case root plus escaped nested field path containing dots

## Testing

  ```bash
cargo test -p lance-datafusion
logical_expr::tests::test_field_path_to_expr
cargo test -p lance
test_optimize_append_preserves_case_sensitive_nullable_vector_column
  cargo fmt --all

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Improved query parsing to correctly preserve case sensitivity for root
column names and handle escaped nested field paths (including dots
inside escaped segments).
* Fixed an issue where appending data and optimizing indexes could leave
unindexed fragments, affecting vector scans.
* Vector search results now remain consistent after append + index
optimization when using case-sensitive nullable vector columns.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: skyshineb

rust/lance-datafusion/src/logical_expr.rs

@@ -281,8 +281,10 @@ pub fn field_path_to_expr(field_path: &str) -> Result<Expr> {
         )));
     }
 
-    // Build the column expression, handling nested fields
-    let mut expr = col(&parts[0]);
+    // Build the column expression, handling nested fields.
+    let mut expr = Expr::Column(datafusion::common::Column::new_unqualified(
+        parts[0].clone(),
+    ));
     for part in &parts[1..] {
         expr = expr.field_newstyle(part);
     }
@@ -297,8 +299,26 @@ mod tests {
     use super::*;
 
     use arrow_schema::{Field, Schema as ArrowSchema};
+    use datafusion::common::Column;
     use datafusion_functions::core::expr_ext::FieldAccessor;
 
+    #[test]
+    fn test_field_path_to_expr_preserves_case_sensitive_root_column() {
+        let expr = field_path_to_expr("VECTOR").unwrap();
+
+        assert_eq!(expr, Expr::Column(Column::new_unqualified("VECTOR")));
+    }
+
+    #[test]
+    fn test_field_path_to_expr_preserves_case_sensitive_escaped_nested_path() {
+        let expr = field_path_to_expr("Parent.`Child.With.Dot`").unwrap();
+
+        assert_eq!(
+            expr,
+            Expr::Column(Column::new_unqualified("Parent")).field_newstyle("Child.With.Dot")
+        );
+    }
+
     #[test]
     fn test_resolve_large_utf8() {
         let arrow_schema = ArrowSchema::new(vec![Field::new("a", DataType::LargeUtf8, false)]);

rust/lance/src/index/append.rs

@@ -863,8 +863,10 @@ mod tests {
     use arrow::datatypes::{Float32Type, UInt32Type};
     use arrow_array::cast::AsArray;
     use arrow_array::{
-        FixedSizeListArray, Int32Array, RecordBatch, RecordBatchIterator, StringArray, UInt32Array,
+        ArrayRef, FixedSizeListArray, Int32Array, RecordBatch, RecordBatchIterator, StringArray,
+        UInt32Array,
     };
+    use arrow_buffer::{BooleanBufferBuilder, NullBuffer};
     use arrow_schema::{DataType, Field, Schema};
     use futures::TryStreamExt;
     use lance_arrow::FixedSizeListArrayExt;
@@ -1005,6 +1007,118 @@ mod tests {
         assert_eq!(num_rows, 2000);
     }
 
+    #[tokio::test]
+    async fn test_optimize_append_preserves_case_sensitive_nullable_vector_column() {
+        const DIM: usize = 64;
+        const ROWS: usize = 1000;
+
+        fn make_vectors(rows: usize, dim: usize, include_null: bool) -> FixedSizeListArray {
+            if include_null {
+                let mut nulls_builder = BooleanBufferBuilder::new(rows);
+                for row_idx in 0..rows {
+                    nulls_builder.append(row_idx != 0);
+                }
+                let nulls = NullBuffer::new(nulls_builder.finish());
+                FixedSizeListArray::try_new(
+                    Arc::new(Field::new("item", DataType::Float32, true)),
+                    dim as i32,
+                    Arc::new(generate_random_array(rows * dim)),
+                    Some(nulls),
+                )
+                .unwrap()
+            } else {
+                FixedSizeListArray::try_new_from_values(
+                    generate_random_array(rows * dim),
+                    dim as i32,
+                )
+                .unwrap()
+            }
+        }
+
+        fn make_batch(
+            schema: Arc<Schema>,
+            start_id: u32,
+            vectors: Arc<FixedSizeListArray>,
+        ) -> RecordBatch {
+            let columns: Vec<ArrayRef> = vec![
+                Arc::new(UInt32Array::from_iter_values(
+                    start_id..start_id + ROWS as u32,
+                )) as ArrayRef,
+                vectors as ArrayRef,
+            ];
+            RecordBatch::try_new(schema, columns).unwrap()
+        }
+
+        let test_dir = TempStrDir::default();
+        let test_uri = test_dir.as_str();
+        let vector_type = DataType::FixedSizeList(
+            Arc::new(Field::new("item", DataType::Float32, true)),
+            DIM as i32,
+        );
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::UInt32, false),
+            Field::new("VECTOR", vector_type, true),
+        ]));
+
+        let initial_vectors = Arc::new(make_vectors(ROWS, DIM, false));
+        let initial_batch = make_batch(schema.clone(), 0, initial_vectors);
+        let batches = RecordBatchIterator::new(std::iter::once(Ok(initial_batch)), schema.clone());
+        let mut dataset = Dataset::write(batches, test_uri, None).await.unwrap();
+
+        let params = VectorIndexParams::with_ivf_pq_params(
+            MetricType::L2,
+            IvfBuildParams::new(2),
+            PQBuildParams {
+                num_sub_vectors: 2,
+                ..Default::default()
+            },
+        );
+        dataset
+            .create_index(&["VECTOR"], IndexType::Vector, None, &params, true)
+            .await
+            .unwrap();
+
+        let appended_vectors = Arc::new(make_vectors(ROWS, DIM, true));
+        let query = appended_vectors.value(5);
+        let appended_batch = make_batch(schema.clone(), ROWS as u32, appended_vectors);
+        let batches = RecordBatchIterator::new(std::iter::once(Ok(appended_batch)), schema);
+        dataset.append(batches, None).await.unwrap();
+
+        let index_name = dataset.load_indices().await.unwrap()[0].name.clone();
+        assert!(
+            !dataset
+                .unindexed_fragments(&index_name)
+                .await
+                .unwrap()
+                .is_empty()
+        );
+
+        dataset
+            .optimize_indices(&OptimizeOptions::append())
+            .await
+            .unwrap();
+
+        let dataset = DatasetBuilder::from_uri(test_uri).load().await.unwrap();
+        assert!(
+            dataset
+                .unindexed_fragments(&index_name)
+                .await
+                .unwrap()
+                .is_empty()
+        );
+
+        let mut scanner = dataset.scan();
+        scanner
+            .nearest("VECTOR", query.as_primitive::<Float32Type>(), 10)
+            .unwrap();
+        let results = scanner.try_into_batch().await.unwrap();
+        assert_eq!(
+            results.num_rows(),
+            10,
+            "expected the requested k=10 nearest-neighbor results"
+        );
+    }
+
     /// Regression: a second `OptimizeOptions::append()` call on a steady-state
     /// vector index used to fall through to `optimize_vector_indices` and write
     /// a new UUID directory + manifest even though nothing had changed. The