fix: resolution of complex type variants in Avro unions

[mzabaluev]

Which issue does this PR close?

• Closes Avro schema resolution not properly supported for complex types #9336

Rationale for this change

When an Avro reader schema has a union type that needs to be resolved against the type in the writer schema, resolution information other than primitive type promotions is not properly handled when creating the decoder.
For example, when the reader schema has a nullable record field that has an added nested field on top of the fields defined in the writer schema, the record type resolution needs to be applied, using a projection with the default field value.

What changes are included in this PR?

Extend the union resolution information in the decoder with variant
data for enum remapping and record projection. The Projector data
structure with Skipper decoders makes part of this information,
which necessitated some refactoring.

Are these changes tested?

TODO:

• Debug failing tests including a busy-loop failure mode.
• Add more unit tests exercising the complex resolutions.

Are there any user-facing changes?

No.

[jecsand838]

@mzabaluev LMGTM!

Overall these changes are extremely clean. Just left a few comments.

[jecsand838]

This new panic in decoder construction can abort the process for malformed/partial union resolution state. Since this path already returns Result, I think using return AvroError::SchemaError (or ParseError) instead of panicking would be better, so callers can decide how to handle the failure.

[jecsand838]

Same here. The panic on unexpected nullable-union mapping turns a recoverable schema mismatch into a hard crash.

[jecsand838]

Same here as well. We should probably be passing AvroErrors back to the caller instead of potentially crashing imo.

[jecsand838]

Using unreachable! here is potentially unsafe. Probably best to also pass an AvroError back here too. Perhaps it's unreachable right now, but all it takes is for an edge case to occur or some change down the line to create the potential for a crash.

[jecsand838]

Nice! Love the clean up here :)

[jecsand838]

I think there's a risk of a regression here by not using the resolve_type to preserve field level resolution, especially the DefaultValue.

See the test below and let me know what you think.

    #[test]
    fn test_resolve_writer_non_union_record_to_reader_union_preserves_defaults() {
        // Writer: record Inner{a: int}
        // Reader: union [Inner{a: int, b: int default 42}, string]
        // The matching child (Inner) should preserve DefaultValue(Int(42)) on field b.
        let writer = Schema::Complex(ComplexType::Record(Record {
            name: "Inner",
            namespace: None,
            doc: None,
            aliases: vec![],
            fields: vec![AvroFieldSchema {
                name: "a",
                doc: None,
                r#type: mk_primitive(PrimitiveType::Int),
                default: None,
                aliases: vec![],
            }],
            attributes: Attributes::default(),
        }));
        let reader = mk_union(vec![
            Schema::Complex(ComplexType::Record(Record {
                name: "Inner",
                namespace: None,
                doc: None,
                aliases: vec![],
                fields: vec![
                    AvroFieldSchema {
                        name: "a",
                        doc: None,
                        r#type: mk_primitive(PrimitiveType::Int),
                        default: None,
                        aliases: vec![],
                    },
                    AvroFieldSchema {
                        name: "b",
                        doc: None,
                        r#type: mk_primitive(PrimitiveType::Int),
                        default: Some(Value::Number(serde_json::Number::from(42))),
                        aliases: vec![],
                    },
                ],
                attributes: Attributes::default(),
            })),
            mk_primitive(PrimitiveType::String),
        ]);
        let mut maker = Maker::new(false, false);
        let dt = maker
            .make_data_type(&writer, Some(&reader), None)
            .expect("resolution should succeed");
        // Verify the union resolution structure
        let resolved = match dt.resolution.as_ref() {
            Some(ResolutionInfo::Union(u)) => u,
            other => panic!("expected union resolution info, got {other:?}"),
        };
        assert!(!resolved.writer_is_union && resolved.reader_is_union);
        // The matching child (Inner at index 0) should have field b with DefaultValue
        let children = match dt.codec() {
            Codec::Union(children, _, _) => children,
            other => panic!("expected union codec, got {other:?}"),
        };
        let inner_fields = match children[0].codec() {
            Codec::Struct(f) => f,
            other => panic!("expected struct codec for Inner, got {other:?}"),
        };
        assert_eq!(inner_fields.len(), 2);
        assert_eq!(inner_fields[1].name(), "b");
        assert_eq!(
            inner_fields[1].data_type().resolution,
            Some(ResolutionInfo::DefaultValue(AvroLiteral::Int(42))),
            "field b should have DefaultValue(Int(42)) from schema resolution"
        );
    }