AVRO-3649: default value for union type should match any type inside the union type.

lang/rust/avro/src/error.rs

@@ -199,6 +199,9 @@ pub enum Error {
     #[error("Could not find matching type in union")]
     FindUnionVariant,
 
+    #[error("Union type should not be empty")]
+    EmptyUnion,
+
     #[error("Array({expected:?}) expected, got {other:?}")]
     GetArray {
         expected: SchemaKind,
@@ -229,7 +232,7 @@ pub enum Error {
     #[error("Unions cannot contain duplicate types")]
     GetUnionDuplicate,
 
-    #[error("Union's first type {0:?} must match the `default`'s value type {1:?}")]
+    #[error("One union type {0:?} must match the `default`'s value type {1:?}")]
     GetDefaultUnion(SchemaKind, ValueKind),
 
     #[error("JSON value {0} claims to be u64 but cannot be converted")]

lang/rust/avro/src/schema.rs

@@ -1385,19 +1385,19 @@ impl Parser {
             .and_then(|schemas| {
                 if let Some(default_value) = default.cloned() {
                     let avro_value = types::Value::from(default_value);
-                    let first_schema = schemas.first();
-                    if let Some(schema) = first_schema {
-                        // Try to resolve the schema
-                        let resolved_value = avro_value.to_owned().resolve(schema);
-                        match resolved_value {
-                            Ok(_) => {}
-                            Err(_) => {
-                                return Err(Error::GetDefaultUnion(
-                                    SchemaKind::from(schema),
-                                    types::ValueKind::from(avro_value),
-                                ));
-                            }
-                        }
+                    let resolved = schemas
+                        .iter()
+                        .any(|schema| avro_value.to_owned().resolve(schema).is_ok());
+
+                    if !resolved {
+                        let schema: Option<&Schema> = schemas.get(0);
+                        return match schema {
+                            Some(first_schema) => Err(Error::GetDefaultUnion(
+                                SchemaKind::from(first_schema),
+                                types::ValueKind::from(avro_value),
+                            )),
+                            None => Err(Error::EmptyUnion),
+                        };
                     }
                 }
                 Ok(schemas)
@@ -4227,4 +4227,40 @@ mod tests {
             _ => panic!("Expected Schema::Record"),
         }
     }
+
+    #[test]
+    fn avro_3649_default_notintfirst() {
+        let schema_str = String::from(
+            r#"
+            {
+                "type": "record",
+                "name": "union_schema_test",
+                "fields": [
+                    {"name": "a", "type": ["string", "int"], "default": 123}
+                ]
+            }
+        "#,
+        );
+
+        let schema = Schema::parse_str(&schema_str).unwrap();
+
+        match schema {
+            Schema::Record { name, fields, .. } => {
+                assert_eq!(name, Name::new("union_schema_test").unwrap());
+                assert_eq!(fields.len(), 1);
+                let field = &fields[0];
+                assert_eq!(&field.name, "a");
+                assert_eq!(&field.default, &Some(json!(123)));
+                match &field.schema {
+                    Schema::Union(union) => {
+                        assert_eq!(union.variants().len(), 2);
+                        assert_eq!(union.variants()[0], Schema::String);
+                        assert_eq!(union.variants()[1], Schema::Int);
+                    }
+                    _ => panic!("Expected Schema::Union"),
+                }
+            }
+            _ => panic!("Expected Schema::Record"),
+        }
+    }
 }

lang/rust/avro/tests/schema.rs

@@ -148,19 +148,19 @@ const UNION_EXAMPLES: &[(&str, bool)] = &[
     ),
     (
         r#"{"name": "foo", "type": ["string", "long"], "default": 1}"#,
-        false,
+        true,
     ),
     (
         r#"{"name": "foo", "type": ["string", "null"], "default": null}"#,
-        false,
+        true,
     ),
     (
         r#"{"name": "foo", "type": ["null", "string"], "default": "null"}"#,
-        false,
+        true,
     ),
     (
         r#"{"name": "foo", "type": ["long", "string"], "default": "str"}"#,
-        false,
+        true,
     ),
 ];