fix: union reader schema resolution (#358)

fix: resolve composite schema for Union reader

schema_compatibility.go

@@ -292,6 +292,11 @@ func (c *SchemaCompatibility) resolve(reader, writer Schema) (schema Schema, res
 	if writer.Type() != reader.Type() {
 		if reader.Type() == Union {
 			for _, schema := range reader.(*UnionSchema).Types() {
+				// Compatibility is not guaranteed for every Union reader schema.
+				// Therefore, we need to check compatibility in every iteration.
+				if err := c.compatible(schema, writer); err != nil {
+					continue
+				}
 				sch, _, err := c.resolve(schema, writer)
 				if err != nil {
 					continue

schema_compatibility_test.go

@@ -862,3 +862,44 @@ func TestSchemaCompatibility_ResolveWithRefs(t *testing.T) {
 	want := map[string]any{"a": "foo"}
 	assert.Equal(t, want, result)
 }
+
+func TestSchemaCompatibility_ResolveWithComplexUnion(t *testing.T) {
+	r := avro.MustParse(`[
+				{
+					"type":"record",
+					"name":"testA",
+					"aliases": ["test1"], 
+					"namespace": "org.hamba.avro", 
+					"fields":[{"name": "a", "type": "long"}]
+				},
+				{
+					"type":"record",
+					"name":"testB",
+					"aliases": ["test2"], 
+					"namespace": "org.hamba.avro", 
+					"fields":[{"name": "b", "type": "bytes"}]
+				}
+			]`)
+
+	w := avro.MustParse(`{
+				"type":"record",
+				"name":"test2",
+				"namespace": "org.hamba.avro", 
+				"fields":[{"name": "b", "type": "string"}]
+			}`)
+
+	value := map[string]any{"b": "foo"}
+	b, err := avro.Marshal(w, value)
+	assert.NoError(t, err)
+
+	sc := avro.NewSchemaCompatibility()
+	sch, err := sc.Resolve(r, w)
+	assert.NoError(t, err)
+
+	var result any
+	err = avro.Unmarshal(sch, b, &result)
+	assert.NoError(t, err)
+
+	want := map[string]any{"b": []byte("foo")}
+	assert.Equal(t, want, result)
+}