fix(spec): preserve exact schema and temporal value semantics

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

@@ -36,7 +36,7 @@ use self::_serde::SchemaEnum;
 use self::id_reassigner::ReassignFieldIds;
 use self::index::{IndexByName, index_by_id, index_parents};
 pub use self::prune_columns::prune_columns;
-use super::NestedField;
+use super::{Literal, NestedField};
 use crate::error::Result;
 use crate::expr::accessor::StructAccessor;
 use crate::spec::FormatVersion;
@@ -71,7 +71,7 @@ pub struct Schema {
     id_to_field: HashMap<i32, NestedFieldRef>,
 
     name_to_id: HashMap<String, i32>,
-    lowercase_name_to_id: HashMap<String, i32>,
+    lowercase_name_to_id: HashMap<String, Option<i32>>,
     id_to_name: HashMap<i32, String>,
 
     field_id_to_accessor: HashMap<i32, Arc<StructAccessor>>,
@@ -150,10 +150,18 @@ 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<String, Option<i32>> =
+            HashMap::with_capacity(name_to_id.len());
+        for (name, id) in &name_to_id {
+            lowercase_name_to_id
+                .entry(name.to_lowercase())
+                .and_modify(|existing| {
+                    if existing.is_some_and(|existing_id| existing_id != *id) {
+                        *existing = None;
+                    }
+                })
+                .or_insert(Some(*id));
+        }
 
         let highest_field_id = id_to_field.keys().max().cloned().unwrap_or(0);
 
@@ -354,7 +362,9 @@ 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))
+            .copied()
+            .flatten()
+            .and_then(|id| self.field_by_id(id))
     }
 
     /// Get field by alias.
@@ -405,8 +415,8 @@ impl Schema {
 
     /// Check if this schema is identical to another schema semantically - excluding schema id.
     pub(crate) fn is_same_schema(&self, other: &SchemaRef) -> bool {
-        self.as_struct().eq(other.as_struct())
-            && self.identifier_field_ids().eq(other.identifier_field_ids())
+        same_struct(self.as_struct(), other.as_struct())
+            && self.identifier_field_ids == other.identifier_field_ids
     }
 
     /// Change the schema id of this schema.
@@ -511,6 +521,48 @@ impl Schema {
     }
 }
 
+fn same_struct(left: &StructType, right: &StructType) -> bool {
+    left.fields().len() == right.fields().len()
+        && left
+            .fields()
+            .iter()
+            .zip(right.fields())
+            .all(|(left, right)| same_field(left, right))
+}
+
+fn same_field(left: &NestedField, right: &NestedField) -> bool {
+    left.id == right.id
+        && left.name == right.name
+        && left.required == right.required
+        && left.doc == right.doc
+        && same_type(&left.field_type, &right.field_type)
+        && same_default(&left.initial_default, &right.initial_default)
+        && same_default(&left.write_default, &right.write_default)
+}
+
+fn same_type(left: &Type, right: &Type) -> bool {
+    match (left, right) {
+        (Type::Primitive(left), Type::Primitive(right)) => left == right,
+        (Type::Struct(left), Type::Struct(right)) => same_struct(left, right),
+        (Type::List(left), Type::List(right)) => {
+            same_field(&left.element_field, &right.element_field)
+        }
+        (Type::Map(left), Type::Map(right)) => {
+            same_field(&left.key_field, &right.key_field)
+                && same_field(&left.value_field, &right.value_field)
+        }
+        (Type::Variant(left), Type::Variant(right)) => left == right,
+        _ => false,
+    }
+}
+
+fn same_default(left: &Option<Literal>, right: &Option<Literal>) -> bool {
+    match (left, right) {
+        (Some(Literal::Primitive(left)), Some(Literal::Primitive(right))) => left.same_value(right),
+        _ => left == right,
+    }
+}
+
 impl Display for Schema {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         writeln!(f, "table {{")?;
@@ -969,6 +1021,21 @@ table {
         }
     }
 
+    #[test]
+    fn test_schema_case_insensitive_lookup_rejects_ambiguous_names() {
+        let schema = Schema::builder()
+            .with_fields([
+                NestedField::optional(1, "foo", Primitive(PrimitiveType::Int)).into(),
+                NestedField::optional(2, "FOO", Primitive(PrimitiveType::Long)).into(),
+            ])
+            .build()
+            .unwrap();
+
+        assert_eq!(schema.field_by_name("foo").unwrap().id, 1);
+        assert_eq!(schema.field_by_name("FOO").unwrap().id, 2);
+        assert!(schema.field_by_name_case_insensitive("fOo").is_none());
+    }
+
     #[test]
     fn test_schema_find_column_name() {
         let expected_column_name = HashMap::from([
@@ -1307,6 +1374,54 @@ table {
         assert_eq!(0, schema.highest_field_id());
     }
 
+    #[test]
+    fn test_same_schema_ignores_identifier_set_iteration_order() {
+        let fields = || {
+            (1..=8)
+                .map(|id| {
+                    NestedField::required(id, format!("id_{id}"), Primitive(PrimitiveType::Long))
+                        .into()
+                })
+                .collect::<Vec<_>>()
+        };
+        let left = Schema::builder()
+            .with_fields(fields())
+            .with_identifier_field_ids(1..=8)
+            .build()
+            .unwrap();
+        let right = Schema::builder()
+            .with_fields(fields())
+            .with_identifier_field_ids((1..=8).rev())
+            .build()
+            .unwrap();
+
+        assert!(left.is_same_schema(&std::sync::Arc::new(right)));
+    }
+
+    #[test]
+    fn test_same_schema_uses_java_floating_point_equality_for_defaults() {
+        let schema_with_default = |default| {
+            Schema::builder()
+                .with_fields([
+                    NestedField::optional(1, "value", Primitive(PrimitiveType::Float))
+                        .with_initial_default(default)
+                        .into(),
+                ])
+                .build()
+                .unwrap()
+        };
+
+        let negative_zero = schema_with_default(Literal::float(-0.0));
+        let positive_zero = std::sync::Arc::new(schema_with_default(Literal::float(0.0)));
+        assert!(!negative_zero.is_same_schema(&positive_zero));
+
+        let first_nan = schema_with_default(Literal::float(f32::from_bits(0x7fc0_0001)));
+        let second_nan = std::sync::Arc::new(schema_with_default(Literal::float(f32::from_bits(
+            0x7fc0_0002,
+        ))));
+        assert!(first_nan.is_same_schema(&second_nan));
+    }
+
     #[test]
     fn test_field_ids_must_be_unique() {
         let reassigned_schema = Schema::builder()

crates/iceberg/src/spec/values/datum.rs

@@ -299,16 +299,16 @@ impl Display for Datum {
             (_, PrimitiveLiteral::Float(val)) => write!(f, "{val}"),
             (_, PrimitiveLiteral::Double(val)) => write!(f, "{val}"),
             (PrimitiveType::Date, PrimitiveLiteral::Int(val)) => {
-                write!(f, "{}", date::days_to_date(*val))
+                write!(f, "{}", date::days_to_iso_date(*val))
             }
             (PrimitiveType::Time, PrimitiveLiteral::Long(val)) => {
                 write!(f, "{}", time::microseconds_to_time(*val))
             }
             (PrimitiveType::Timestamp, PrimitiveLiteral::Long(val)) => {
-                write!(f, "{}", timestamp::microseconds_to_datetime(*val))
+                write!(f, "{}", timestamp::microseconds_to_display(*val))
             }
             (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(val)) => {
-                write!(f, "{}", timestamptz::microseconds_to_datetimetz(*val))
+                write!(f, "{}", timestamptz::microseconds_to_display(*val))
             }
             (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(val)) => {
                 write!(f, "{}", timestamp::nanoseconds_to_datetime(*val))

crates/iceberg/src/spec/values/literal.rs

@@ -460,11 +460,9 @@ impl Literal {
                         "Failed to convert json number to double",
                     ))?)),
                 ))),
-                (PrimitiveType::Date, JsonValue::String(s)) => {
-                    Ok(Some(Literal::Primitive(PrimitiveLiteral::Int(
-                        date::date_to_days(&NaiveDate::parse_from_str(&s, "%Y-%m-%d")?),
-                    ))))
-                }
+                (PrimitiveType::Date, JsonValue::String(s)) => Ok(Some(Literal::Primitive(
+                    PrimitiveLiteral::Int(date::iso_date_to_days(&s)?),
+                ))),
                 (PrimitiveType::Date, JsonValue::Number(number)) => {
                     Ok(Some(Literal::Primitive(PrimitiveLiteral::Int(
                         number
@@ -482,17 +480,11 @@ impl Literal {
                     ))))
                 }
                 (PrimitiveType::Timestamp, JsonValue::String(s)) => Ok(Some(Literal::Primitive(
-                    PrimitiveLiteral::Long(timestamp::datetime_to_microseconds(
-                        &NaiveDateTime::parse_from_str(&s, "%Y-%m-%dT%H:%M:%S%.f")?,
-                    )),
+                    PrimitiveLiteral::Long(timestamp::iso_datetime_to_microseconds(&s)?),
+                ))),
+                (PrimitiveType::Timestamptz, JsonValue::String(s)) => Ok(Some(Literal::Primitive(
+                    PrimitiveLiteral::Long(timestamptz::iso_datetime_to_microseconds(&s)?),
                 ))),
-                (PrimitiveType::Timestamptz, JsonValue::String(s)) => {
-                    Ok(Some(Literal::Primitive(PrimitiveLiteral::Long(
-                        timestamptz::datetimetz_to_microseconds(&Utc.from_utc_datetime(
-                            &NaiveDateTime::parse_from_str(&s, "%Y-%m-%dT%H:%M:%S%.f+00:00")?,
-                        )),
-                    ))))
-                }
                 (PrimitiveType::TimestampNs, JsonValue::String(s)) => {
                     let ndt = NaiveDateTime::parse_from_str(&s, "%Y-%m-%dT%H:%M:%S%.f")?;
                     let nanos = timestamp::datetime_to_nanoseconds(&ndt).ok_or_else(|| {
@@ -663,20 +655,16 @@ impl Literal {
                     }
                 }
                 (PrimitiveType::Date, PrimitiveLiteral::Int(val)) => {
-                    Ok(JsonValue::String(date::days_to_date(val).to_string()))
+                    Ok(JsonValue::String(date::days_to_iso_date(val)))
                 }
                 (PrimitiveType::Time, PrimitiveLiteral::Long(val)) => Ok(JsonValue::String(
                     time::microseconds_to_time(val).to_string(),
                 )),
                 (PrimitiveType::Timestamp, PrimitiveLiteral::Long(val)) => Ok(JsonValue::String(
-                    timestamp::microseconds_to_datetime(val)
-                        .format("%Y-%m-%dT%H:%M:%S%.f")
-                        .to_string(),
+                    timestamp::microseconds_to_iso_datetime(val),
                 )),
                 (PrimitiveType::Timestamptz, PrimitiveLiteral::Long(val)) => Ok(JsonValue::String(
-                    timestamptz::microseconds_to_datetimetz(val)
-                        .format("%Y-%m-%dT%H:%M:%S%.f+00:00")
-                        .to_string(),
+                    timestamptz::microseconds_to_iso_datetime(val),
                 )),
                 (PrimitiveType::TimestampNs, PrimitiveLiteral::Long(val)) => Ok(JsonValue::String(
                     timestamp::nanoseconds_to_datetime(val)

crates/iceberg/src/spec/values/primitive.rs

@@ -47,6 +47,18 @@ pub enum PrimitiveLiteral {
 }
 
 impl PrimitiveLiteral {
+    pub(crate) fn same_value(&self, other: &Self) -> bool {
+        match (self, other) {
+            (PrimitiveLiteral::Float(left), PrimitiveLiteral::Float(right)) => {
+                (left.is_nan() && right.is_nan()) || left.0.to_bits() == right.0.to_bits()
+            }
+            (PrimitiveLiteral::Double(left), PrimitiveLiteral::Double(right)) => {
+                (left.is_nan() && right.is_nan()) || left.0.to_bits() == right.0.to_bits()
+            }
+            _ => self == other,
+        }
+    }
+
     /// Returns true if the Literal represents a primitive type
     /// that can be a NaN, and that it's value is NaN
     pub fn is_nan(&self) -> bool {
@@ -57,3 +69,27 @@ impl PrimitiveLiteral {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_same_value_uses_java_floating_point_equality() {
+        assert!(
+            !PrimitiveLiteral::Float(OrderedFloat(-0.0))
+                .same_value(&PrimitiveLiteral::Float(OrderedFloat(0.0)))
+        );
+        assert!(
+            !PrimitiveLiteral::Double(OrderedFloat(-0.0))
+                .same_value(&PrimitiveLiteral::Double(OrderedFloat(0.0)))
+        );
+
+        let first_nan = f32::from_bits(0x7fc0_0001);
+        let second_nan = f32::from_bits(0x7fc0_0002);
+        assert!(
+            PrimitiveLiteral::Float(OrderedFloat(first_nan))
+                .same_value(&PrimitiveLiteral::Float(OrderedFloat(second_nan)))
+        );
+    }
+}