fix: preserve kernel column segments (delta-io#4164)

# Description
Fix kernel to DataFusion column expression conversion to preserve exact
`ColumnName` path segments.

Fixes delta-io#4082

**Fix:**
Use DataFusion `ident(...)` for the base column segment when converting
`Expression::Column`, then `.field(...)` for remaining path segments.
Preserves exact segment names, avoids SQL style normalization.

# Related Issue(s)
- closes delta-io#4082

<!---
For example:

- closes #106
--->

# Documentation

<!---
Share links to useful documentation
--->

---------

Signed-off-by: Ethan Urbanski <ethan@urbanskitech.com>

Author: ethan-tyler

crates/core/src/delta_datafusion/engine/expressions/to_datafusion.rs

@@ -7,7 +7,7 @@ use datafusion::common::{
 use datafusion::functions::core::expr_ext::FieldAccessor;
 use datafusion::functions::expr_fn::named_struct;
 use datafusion::logical_expr::expr::ScalarFunction;
-use datafusion::logical_expr::{BinaryExpr, Expr, Operator, col, lit};
+use datafusion::logical_expr::{BinaryExpr, Expr, Operator, expr_fn::ident, lit};
 use datafusion::prelude::coalesce;
 use delta_kernel::Predicate;
 use delta_kernel::arrow::datatypes::{DataType as ArrowDataType, Field as ArrowField};
@@ -34,7 +34,11 @@ pub(crate) fn to_datafusion_expr(expr: &Expression, output_type: &DataType) -> D
             let base_name = name_iter
                 .next()
                 .ok_or_else(|| internal_datafusion_err!("Expected at least one column name"))?;
-            Ok(name_iter.fold(col(base_name), |acc, n| acc.field(n)))
+            // Kernel column paths are already exact segments. Using `ident` avoids DataFusion SQL
+            // identifier normalization (e.g. lowercasing), and treats `.` as a literal character
+            // rather than a qualifier separator. This is required for correctness with camelCase
+            // and other non-normalized column names (see #4082).
+            Ok(name_iter.fold(ident(base_name), |acc, n| acc.field(n)))
         }
         Expression::Predicate(expr) => predicate_to_df(expr, output_type),
         Expression::Struct(fields) => struct_to_df(fields, output_type),
@@ -222,7 +226,7 @@ fn struct_to_df(fields: &[Arc<Expression>], output_type: &DataType) -> DFResult<
 mod tests {
     use std::ops::Not;
 
-    use datafusion::logical_expr::{col, lit};
+    use datafusion::logical_expr::{col, expr_fn::ident, lit};
     use delta_kernel::expressions::ColumnName;
     use delta_kernel::expressions::{ArrayData, BinaryExpression, MapData, Scalar, StructData};
     use delta_kernel::schema::{ArrayType, DataType, MapType, StructField, StructType};
@@ -381,11 +385,25 @@ mod tests {
     fn test_column_expression() {
         let expr = Expression::Column(ColumnName::new(["test_col"]));
         let result = to_datafusion_expr(&expr, &DataType::BOOLEAN).unwrap();
-        assert_eq!(result, col("test_col"));
+        assert_eq!(result, ident("test_col"));
 
         let expr = Expression::Column(ColumnName::new(["test_col", "field_1", "field_2"]));
         let result = to_datafusion_expr(&expr, &DataType::BOOLEAN).unwrap();
-        assert_eq!(result, col("test_col").field("field_1").field("field_2"));
+        assert_eq!(result, ident("test_col").field("field_1").field("field_2"));
+    }
+
+    #[test]
+    fn test_column_expression_preserves_case() {
+        let expr = Expression::Column(ColumnName::new(["submittedAt"]));
+        let result = to_datafusion_expr(&expr, &DataType::BOOLEAN).unwrap();
+        assert_eq!(result, ident("submittedAt"));
+    }
+
+    #[test]
+    fn test_column_expression_preserves_dots_in_segments() {
+        let expr = Expression::Column(ColumnName::new(["a.b"]));
+        let result = to_datafusion_expr(&expr, &DataType::BOOLEAN).unwrap();
+        assert_eq!(result, ident("a.b"));
     }
 
     /// Test various literal values:

python/tests/test_merge.py

@@ -2117,6 +2117,48 @@ def test_merge_preserves_casing_in_quoted_identifiers(
     assert result == expected
 
 
+def test_merge_camelcase_non_nullable_column_4082(tmp_path: pathlib.Path):
+    # Regression test for https://github.com/delta-io/delta-rs/issues/4082
+    schema = Schema(
+        [
+            Field("submittedAt", PrimitiveType("long"), nullable=False),
+            Field("id", PrimitiveType("string"), nullable=True),
+        ]
+    )
+    dt = DeltaTable.create(tmp_path, schema=schema, mode="ignore")
+
+    source_table = Table(
+        {
+            "submittedAt": Array(
+                [123],
+                ArrowField("submittedAt", type=DataType.int64(), nullable=True),
+            ),
+            "id": Array(
+                ["test"],
+                ArrowField("id", type=DataType.string_view(), nullable=True),
+            ),
+        }
+    )
+
+    (
+        dt.merge(
+            source=source_table,
+            predicate="source.id = target.id",
+            source_alias="source",
+            target_alias="target",
+        )
+        .when_matched_update_all()
+        .when_not_matched_insert_all()
+        .execute()
+    )
+
+    assert dt.history(1)[0]["operation"] == "MERGE"
+
+    result = QueryBuilder().register("tbl", dt).execute("select * from tbl").read_all()
+    assert result["submittedAt"].to_pylist() == [123]
+    assert result["id"].to_pylist() == ["test"]
+
+
 @pytest.mark.pandas
 @pytest.mark.pyarrow
 def test_struct_casting(tmp_path: pathlib.Path):