fix(scan): reject ambiguous case-insensitive fields

crates/iceberg/src/expr/term.rs

@@ -311,10 +311,10 @@ impl Bind for Reference {
 
     fn bind(&self, schema: SchemaRef, case_sensitive: bool) -> crate::Result<Self::Bound> {
         let field = if case_sensitive {
-            schema.field_by_name(&self.name)
+            Ok(schema.field_by_name(&self.name))
         } else {
-            schema.field_by_name_case_insensitive(&self.name)
-        };
+            schema.field_by_name_case_insensitive_checked(&self.name)
+        }?;
 
         let field = field.ok_or_else(|| {
             Error::new(

crates/iceberg/src/scan/mod.rs

@@ -51,13 +51,17 @@ use crate::{Error, ErrorKind, Result};
 pub type ArrowRecordBatchStream = BoxStream<'static, Result<RecordBatch>>;
 
 /// Resolves a column name to its field ID, honouring the scan's case sensitivity.
-fn resolve_field_id(schema: &Schema, column_name: &str, case_sensitive: bool) -> Option<i32> {
+fn resolve_field_id(
+    schema: &Schema,
+    column_name: &str,
+    case_sensitive: bool,
+) -> Result<Option<i32>> {
     if case_sensitive {
-        schema.field_id_by_name(column_name)
+        Ok(schema.field_id_by_name(column_name))
     } else {
-        schema
-            .field_by_name_case_insensitive(column_name)
-            .map(|field| field.id)
+        Ok(schema
+            .field_by_name_case_insensitive_checked(column_name)?
+            .map(|field| field.id))
     }
 }
 
@@ -251,8 +255,8 @@ impl<'a> TableScanBuilder<'a> {
                 continue;
             }
 
-            let field_id =
-                resolve_field_id(&schema, column_name, self.case_sensitive).ok_or_else(|| {
+            let field_id = resolve_field_id(&schema, column_name, self.case_sensitive)?
+                .ok_or_else(|| {
                     Error::new(
                         ErrorKind::DataInvalid,
                         format!("Column {column_name} not found in table. Schema: {schema}"),
@@ -653,8 +657,9 @@ pub mod tests {
     use tempfile::TempDir;
     use uuid::Uuid;
 
+    use super::resolve_field_id;
     use crate::arrow::ArrowReaderBuilder;
-    use crate::expr::{BoundPredicate, Reference};
+    use crate::expr::{Bind, BoundPredicate, Reference};
     use crate::io::{FileIO, OutputFile};
     use crate::metadata_columns::{
         RESERVED_COL_NAME_DELETE_FILE_PATH, RESERVED_COL_NAME_DELETE_FILE_POS,
@@ -1780,6 +1785,29 @@ pub mod tests {
         );
     }
 
+    #[test]
+    fn test_case_insensitive_scan_rejects_ambiguous_column_name() {
+        let schema = Schema::builder()
+            .with_fields(vec![
+                NestedField::optional(1, "id", PrimitiveType::Int.into()).into(),
+                NestedField::optional(2, "ID", PrimitiveType::Int.into()).into(),
+            ])
+            .build()
+            .unwrap();
+
+        assert_eq!(resolve_field_id(&schema, "id", true).unwrap(), Some(1));
+        let error = resolve_field_id(&schema, "Id", false).unwrap_err();
+        assert_eq!(error.kind(), ErrorKind::DataInvalid);
+        assert!(error.to_string().contains("ambiguous"), "{error}");
+
+        let error = Reference::new("Id")
+            .is_null()
+            .bind(Arc::new(schema), false)
+            .unwrap_err();
+        assert_eq!(error.kind(), ErrorKind::DataInvalid);
+        assert!(error.to_string().contains("ambiguous"), "{error}");
+    }
+
     #[tokio::test]
     async fn test_case_insensitive_scan_resolves_mismatched_column_case() {
         let mut fixture = TableTestFixture::new();

crates/iceberg/src/spec/schema/mod.rs

@@ -71,7 +71,8 @@ pub struct Schema {
     id_to_field: HashMap<i32, NestedFieldRef>,
 
     name_to_id: HashMap<String, i32>,
-    lowercase_name_to_id: HashMap<String, i32>,
+    // `None` marks a lower-cased name that resolves to more than one field.
+    lowercase_name_to_id: HashMap<String, Option<i32>>,
     id_to_name: HashMap<i32, String>,
 
     field_id_to_accessor: HashMap<i32, Arc<StructAccessor>>,
@@ -150,10 +151,13 @@ impl SchemaBuilder {
             index.indexes()
         };
 
-        let lowercase_name_to_id = name_to_id
-            .iter()
-            .map(|(k, v)| (k.to_lowercase(), *v))
-            .collect();
+        let mut lowercase_name_to_id = HashMap::new();
+        for (name, field_id) in &name_to_id {
+            lowercase_name_to_id
+                .entry(name.to_lowercase())
+                .and_modify(|existing| *existing = None)
+                .or_insert(Some(*field_id));
+        }
 
         let highest_field_id = id_to_field.keys().max().cloned().unwrap_or(0);
 
@@ -354,7 +358,21 @@ impl Schema {
     pub fn field_by_name_case_insensitive(&self, field_name: &str) -> Option<&NestedFieldRef> {
         self.lowercase_name_to_id
             .get(&field_name.to_lowercase())
-            .and_then(|id| self.field_by_id(*id))
+            .and_then(|id| id.and_then(|id| self.field_by_id(id)))
+    }
+
+    pub(crate) fn field_by_name_case_insensitive_checked(
+        &self,
+        field_name: &str,
+    ) -> Result<Option<&NestedFieldRef>> {
+        match self.lowercase_name_to_id.get(&field_name.to_lowercase()) {
+            Some(Some(id)) => Ok(self.field_by_id(*id)),
+            Some(None) => Err(Error::new(
+                ErrorKind::DataInvalid,
+                format!("Field name {field_name} is ambiguous when case sensitivity is disabled"),
+            )),
+            None => Ok(None),
+        }
     }
 
     /// Get field by alias.