fix(planner): preserve physical column casing in graph scans

crates/lance-csr/src/datafusion_planner/join_ops.rs

@@ -8,11 +8,13 @@
 //! join-key construction, filter push-downs, and aliasing conventions so that
 //! higher-level planner code can remain focused on traversal semantics.
 
+use super::scan_ops::source_column;
 use super::DataFusionPlanner;
 use crate::ast::{PropertyValue, RelationshipDirection};
 use crate::case_insensitive::qualify_column;
 use crate::config::{NodeMapping, RelationshipMapping};
 use crate::error::Result;
+use datafusion::common::Column;
 use datafusion::logical_expr::{
     col, BinaryExpr, Expr, JoinType, LogicalPlan, LogicalPlanBuilder, Operator, TableSource,
 };
@@ -62,7 +64,7 @@ impl DataFusionPlanner {
             let lit_expr =
                 self.unresolved_value_expr(&crate::ast::ValueExpression::Literal(v.clone()));
             let filter_expr = Expr::BinaryExpr(BinaryExpr {
-                left: Box::new(col(k.to_lowercase())),
+                left: Box::new(source_column(&target_schema, k)),
                 op: Operator::Eq,
                 right: Box::new(lit_expr),
             });
@@ -76,7 +78,7 @@ impl DataFusionPlanner {
             .iter()
             .map(|field| {
                 let qualified_name = qualify_column(target_variable, field.name());
-                col(field.name()).alias(&qualified_name)
+                col(Column::new_unqualified(field.name())).alias(&qualified_name)
             })
             .collect();
 

crates/lance-csr/src/datafusion_planner/scan_ops.rs

@@ -10,10 +10,21 @@ use crate::ast::{PropertyValue, RelationshipDirection};
 use crate::case_insensitive::qualify_column;
 use crate::config::RelationshipMapping;
 use crate::error::Result;
+use datafusion::common::Column;
 use datafusion::logical_expr::{col, BinaryExpr, Expr, LogicalPlan, LogicalPlanBuilder, Operator};
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
+/// Bind a property filter to its physical field spelling before scan aliasing.
+pub(super) fn source_column(schema: &arrow_schema::Schema, name: &str) -> Expr {
+    let physical = schema
+        .fields()
+        .iter()
+        .find(|field| field.name().eq_ignore_ascii_case(name))
+        .map_or(name, |field| field.name().as_str());
+    col(Column::new_unqualified(physical))
+}
+
 // Forward declare DataFusionPlanner to add methods to it
 use super::DataFusionPlanner;
 
@@ -48,7 +59,7 @@ impl DataFusionPlanner {
                                 &crate::ast::ValueExpression::Literal(v.clone()),
                             );
                             Expr::BinaryExpr(BinaryExpr {
-                                left: Box::new(col(k)),
+                                left: Box::new(source_column(&schema, k)),
                                 op: Operator::Eq,
                                 right: Box::new(lit_expr),
                             })
@@ -79,7 +90,7 @@ impl DataFusionPlanner {
                     .iter()
                     .map(|field| {
                         let qualified_name = qualify_column(variable, field.name());
-                        col(field.name()).alias(&qualified_name)
+                        col(Column::new_unqualified(field.name())).alias(&qualified_name)
                     })
                     .collect();
 
@@ -143,7 +154,7 @@ impl DataFusionPlanner {
             let lit_expr =
                 self.unresolved_value_expr(&crate::ast::ValueExpression::Literal(v.clone()));
             let filter_expr = Expr::BinaryExpr(BinaryExpr {
-                left: Box::new(col(k)),
+                left: Box::new(source_column(&rel_schema, k)),
                 op: Operator::Eq,
                 right: Box::new(lit_expr),
             });
@@ -163,7 +174,7 @@ impl DataFusionPlanner {
             .iter()
             .map(|field| {
                 let qualified_name = qualify_column(&rel_instance.alias, field.name());
-                col(field.name()).alias(&qualified_name)
+                col(Column::new_unqualified(field.name())).alias(&qualified_name)
             })
             .collect();
 
@@ -209,14 +220,14 @@ impl DataFusionPlanner {
             .fields()
             .iter()
             .map(|f| match f.name() {
-                n if *n == src => col(&dst).alias(&src),
-                n if *n == dst => col(&src).alias(&dst),
-                n => col(n),
+                n if *n == src => col(Column::new_unqualified(&dst)).alias(&src),
+                n if *n == dst => col(Column::new_unqualified(&src)).alias(&dst),
+                n => col(Column::new_unqualified(n)),
             })
             .collect();
         let reversed = builder
             .clone()
-            .filter(col(&src).not_eq(col(&dst)))
+            .filter(col(Column::new_unqualified(&src)).not_eq(col(Column::new_unqualified(&dst))))
             .and_then(|b| b.project(swapped))
             .and_then(|b| b.build())
             .map_err(|e| self.plan_error("Failed to reverse undirected edges", e))?;
@@ -273,6 +284,49 @@ mod tests {
     };
     use crate::logical_plan::{LogicalOperator, ProjectionItem};
 
+    #[tokio::test]
+    async fn test_mixed_case_physical_fields_and_inline_filter() {
+        use arrow_array::{Int64Array, RecordBatch, StringArray};
+        use datafusion::execution::context::SessionContext;
+        use std::sync::Arc;
+
+        let ctx = SessionContext::new();
+        ctx.register_batch(
+            "person",
+            RecordBatch::try_from_iter(vec![
+                ("ID", Arc::new(Int64Array::from(vec![1, 2])) as _),
+                (
+                    "firstName",
+                    Arc::new(StringArray::from(vec!["Bill", "Alice"])) as _,
+                ),
+            ])
+            .unwrap(),
+        )
+        .unwrap();
+        let config = person_config();
+        for query in [
+            "MATCH (p:Person {FIRSTNAME: 'Alice'}) RETURN p.id AS id",
+            "MATCH (p:Person) WHERE p.firstName = 'Alice' RETURN p.ID AS id",
+        ] {
+            let result = crate::query::CypherQuery::new(query)
+                .unwrap()
+                .with_config(config.clone())
+                .execute_with_context(ctx.clone())
+                .await
+                .unwrap();
+            assert_eq!(result.num_rows(), 1);
+            assert_eq!(
+                result
+                    .column(0)
+                    .as_any()
+                    .downcast_ref::<Int64Array>()
+                    .unwrap()
+                    .value(0),
+                2
+            );
+        }
+    }
+
     #[test]
     fn test_df_planner_inline_property_filter() {
         let mut props = std::collections::HashMap::new();

docs/cypher.md

@@ -522,6 +522,11 @@ print(result.to_pylist())
 
 ## Case sensitivity
 
+Mixed-case Arrow and Lance fields follow the same identifier rule: `p.firstName`, `p.firstname`
+and `p.FIRSTNAME` refer to the same property, including inline filters and relationship properties.
+Source field names are preserved. Explicit result aliases use the existing lowercase convention.
+
+
 Keywords, function names, variables, labels, relationship types and property names are all
 case-insensitive. `match (P:person) return p.NAME` works, and the output column keeps the spelling you wrote.
 

python/python/tests/test_column_casing.py

@@ -0,0 +1,119 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The Lance Authors
+
+"""Catalog scans preserve physical spelling while Cypher names ignore case."""
+
+import lance
+import pyarrow as pa
+import pytest
+from lance_csr import CypherEngine, CypherQuery, GraphConfig
+
+
+@pytest.fixture(params=["arrow", "lance"])
+def graph(request, tmp_path):
+    config = (
+        GraphConfig.builder()
+        .node_table("Person", "id")
+        .rel_table("KNOWS", "Person", "Person", "sourceid", "targetid")
+        .build()
+    )
+    tables = {
+        "Person": pa.table(
+            {
+                "ID": [1, 2, 3],
+                "firstName": ["Bill", "Alice", "Carol"],
+                "isMarried": [True, False, None],
+            }
+        ),
+        "KNOWS": pa.table(
+            {"sourceID": [1, 2], "targetID": [2, 3], "createdAt": [2001, 2002]}
+        ),
+    }
+    sources = (
+        tables
+        if request.param == "arrow"
+        else {
+            name: lance.write_dataset(table, tmp_path / f"{name}.lance")
+            for name, table in tables.items()
+        }
+    )
+    return config, tables, sources
+
+
+@pytest.mark.parametrize("mode", ["legacy", "join", "csr", "stream-join", "stream-csr"])
+@pytest.mark.parametrize(
+    ("query", "expected"),
+    [
+        (
+            "MATCH (p:Person) RETURN p.isMarried AS married ORDER BY p.id",
+            [{"married": True}, {"married": False}, {"married": None}],
+        ),
+        (
+            "MATCH (p:PERSON) WHERE p.ISMARRIED = true RETURN p.FIRSTNAME AS name",
+            [{"name": "Bill"}],
+        ),
+        ("MATCH (p:Person {firstName: 'Alice'}) RETURN p.id AS id", [{"id": 2}]),
+        (
+            "MATCH (p:Person) RETURN p.id AS id ORDER BY id",
+            [{"id": 1}, {"id": 2}, {"id": 3}],
+        ),
+        (
+            "MATCH (a:Person)-[r:KNOWS]->(b:Person) "
+            "RETURN r.createdAt AS year ORDER BY year",
+            [{"year": 2001}, {"year": 2002}],
+        ),
+        (
+            "MATCH (a:Person)-[r:KNOWS {CREATEDAT: 2002}]->(b:Person) "
+            "RETURN b.firstName AS name",
+            [{"name": "Carol"}],
+        ),
+        (
+            "MATCH (a:Person)-[:KNOWS]-(b:Person) WHERE a.id = 2 "
+            "RETURN b.id AS id ORDER BY id",
+            [{"id": 1}, {"id": 3}],
+        ),
+        (
+            "MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE a.id = 1 "
+            "RETURN b.firstName AS name",
+            [{"name": "Alice"}],
+        ),
+        (
+            "MATCH (a:Person)-[:KNOWS]->(b:Person {FIRSTNAME: 'Alice'}) "
+            "RETURN a.id AS id",
+            [{"id": 1}],
+        ),
+        (
+            "MATCH (a:Person)-[:KNOWS*1..2]->(b:Person) WHERE a.id = 1 "
+            "RETURN b.firstName AS name ORDER BY name",
+            [{"name": "Alice"}, {"name": "Carol"}],
+        ),
+        (
+            "MATCH (a:Person) WHERE a.id = 3 "
+            "OPTIONAL MATCH (a)-[:KNOWS]->(b:Person) RETURN b.firstName AS name",
+            [{"name": None}],
+        ),
+        (
+            "MATCH (p:Person) WITH p.firstName AS fullName "
+            "RETURN fullName ORDER BY fullName",
+            [{"fullname": "Alice"}, {"fullname": "Bill"}, {"fullname": "Carol"}],
+        ),
+        (
+            "MATCH (p:Person) RETURN count(p.id) AS numPeople ORDER BY numPeople",
+            [{"numpeople": 3}],
+        ),
+    ],
+)
+def test_physical_column_casing(graph, mode, query, expected):
+    config, tables, sources = graph
+    if mode == "legacy":
+        result = CypherQuery(query).with_config(config).execute(tables)
+    else:
+        engine = CypherEngine(config, sources, use_csr="csr" in mode)
+        result = (
+            engine.execute_batches(query).read_all()
+            if mode.startswith("stream")
+            else engine.execute(query)
+        )
+    assert result.to_pylist() == expected
+    assert result.column_names == list(expected[0])
+    assert tables["Person"].column_names == ["ID", "firstName", "isMarried"]