fix(spec): reject unequal map default arrays

crates/iceberg/src/spec/datatypes.rs

@@ -19,7 +19,6 @@
  * Data Types
  */
 use std::collections::HashMap;
-use std::convert::identity;
 use std::fmt;
 use std::ops::Index;
 use std::sync::{Arc, OnceLock};
@@ -554,7 +553,7 @@ impl fmt::Display for StructType {
 }
 
 #[derive(Debug, PartialEq, Serialize, Deserialize, Eq, Clone)]
-#[serde(from = "SerdeNestedField", into = "SerdeNestedField")]
+#[serde(try_from = "SerdeNestedField", into = "SerdeNestedField")]
 /// A struct is a tuple of typed values. Each field in the tuple is named and has an integer id that is unique in the table schema.
 /// Each field can be either optional or required, meaning that values can (or cannot) be null. Fields may be any type.
 /// Fields may have an optional comment or doc string. Fields can have default values.
@@ -591,25 +590,30 @@ struct SerdeNestedField {
     pub write_default: Option<JsonValue>,
 }
 
-impl From<SerdeNestedField> for NestedField {
-    fn from(value: SerdeNestedField) -> Self {
-        NestedField {
+impl TryFrom<SerdeNestedField> for NestedField {
+    type Error = crate::Error;
+
+    fn try_from(value: SerdeNestedField) -> Result<Self> {
+        let initial_default = value
+            .initial_default
+            .map(|x| Literal::try_from_json(x, &value.field_type))
+            .transpose()?
+            .flatten();
+        let write_default = value
+            .write_default
+            .map(|x| Literal::try_from_json(x, &value.field_type))
+            .transpose()?
+            .flatten();
+
+        Ok(NestedField {
             id: value.id,
             name: value.name,
             required: value.required,
-            initial_default: value.initial_default.and_then(|x| {
-                Literal::try_from_json(x, &value.field_type)
-                    .ok()
-                    .and_then(identity)
-            }),
-            write_default: value.write_default.and_then(|x| {
-                Literal::try_from_json(x, &value.field_type)
-                    .ok()
-                    .and_then(identity)
-            }),
+            initial_default,
+            write_default,
             field_type: value.field_type,
             doc: value.doc,
-        }
+        })
     }
 }
 
@@ -1354,6 +1358,31 @@ mod tests {
         assert_eq!(field, roundtrip);
     }
 
+    #[test]
+    fn nested_field_rejects_invalid_map_defaults() {
+        for default_name in ["initial-default", "write-default"] {
+            let json = format!(
+                r#"{{
+                    "id": 1,
+                    "name": "properties",
+                    "required": false,
+                    "type": {{
+                        "type": "map",
+                        "key-id": 2,
+                        "key": "string",
+                        "value-id": 3,
+                        "value-required": false,
+                        "value": "int"
+                    }},
+                    "{default_name}": {{"keys": ["a", "b"], "values": [1]}}
+                }}"#
+            );
+
+            let error = serde_json::from_str::<NestedField>(&json).unwrap_err();
+            assert!(error.to_string().contains("must have the same length"));
+        }
+    }
+
     #[test]
     fn struct_type_with_type_field() {
         // Test that StructType properly deserializes JSON with "type":"struct" field

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

@@ -598,6 +598,17 @@ impl Literal {
                     if let (Some(JsonValue::Array(keys)), Some(JsonValue::Array(values))) =
                         (object.remove("keys"), object.remove("values"))
                     {
+                        if keys.len() != values.len() {
+                            return Err(Error::new(
+                                ErrorKind::DataInvalid,
+                                format!(
+                                    "Map keys and values must have the same length, got {} keys and {} values",
+                                    keys.len(),
+                                    values.len()
+                                ),
+                            ));
+                        }
+
                         Ok(Some(Literal::Map(Map::from_iter(
                             keys.into_iter()
                                 .zip(values)

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

@@ -426,6 +426,25 @@ fn json_map() {
     );
 }
 
+#[test]
+fn json_map_rejects_mismatched_key_value_lengths() {
+    let map_type = Type::Map(MapType {
+        key_field: NestedField::map_key_element(0, Primitive(PrimitiveType::String)).into(),
+        value_field: NestedField::map_value_element(1, Primitive(PrimitiveType::Int), true).into(),
+    });
+
+    for record in [
+        r#"{"keys":["a","b"],"values":[1]}"#,
+        r#"{"keys":["a"],"values":[1,2]}"#,
+    ] {
+        let value = serde_json::from_str::<JsonValue>(record).unwrap();
+        let error = Literal::try_from_json(value, &map_type).unwrap_err();
+
+        assert_eq!(error.kind(), ErrorKind::DataInvalid);
+        assert!(error.to_string().contains("must have the same length"));
+    }
+}
+
 #[test]
 fn avro_bytes_boolean() {
     let bytes = vec![1u8];