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.

[mzabaluev]

I meant to catch the cases where it's clearly a programming error if any other combination occurs from reading the schema, and the graceful failure would be too cryptic and require a debugging session to investigate. But OK.

[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 :)

[mzabaluev]

The borrow checker made me do this, but yes, it's also become cleaner.

[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"
        );
    }

[mzabaluev-flarion]

Since the writer type is not a union in this case, the only resolution we need is with the best matching union variant that is found above.

A case could be made for not processing the variant schemas the second time to build the field information for the codec, but previous code does that as well since the loop in find_best_union_match exits early upon finding the direct match. Since schema resolution is not on a critical stage for performance, I left it to work as before.

[mzabaluev]

You are right, the test fails because inner resolution information is not preserved. The same should be done when resolving the union-to-union case.

[mzabaluev]

The default value should be retrieved when parsing any Avro schema, not only when resolving. The current representation looks like a data model mismatch.

[mzabaluev]

I have tried to address this the best I could, but the fix (making DefaultValue resolutions for record fields in any parsed schema) invites more test failures.

[mzabaluev]

While default values are a property of record fields, other nested resolutions are particular for each writer-to-reader resolution case. If the writer schema is a union with multiple branches resolving to the same reader branch, fields in any nested records may need to be resolved differently depending on the branch encoded. The current data structure does not allow for that.