Schema resolution lacks named type checking and allows invalid coercions

[KitFieldhouse]

Hi all!

While working on issue #353, I have also bumped into some possible issues with how we currently perform writer to reader schema resolution.

There are two issues here:

• Resolution coercions in this implementation that are not in the specification.
• Not using type names in resolution.

Currently, in my reading of the code, writer to reader schema resolution is performed by first deserializing some buffer into a Value using the "writer" schema and then by calling resolve (or resolve_schemata) on this value with the reader schema. I think the fact that we are performing schema resolution on the value directly without reference to the original writer schema is central to both of these issues.

Resolution coercions not in the specification

Currently, we are allowed to resolve Value::Map against a Schema::Record and to resolve Value::Long against a Schema::Int. This last resolution rule causes buggy behavior, as seen by this test.

#[test]
fn test_long_to_int() -> TestResult {
    init();
    let writer_schema = r#"{
        "type": "record",
        "name": "ASchema",
        "fields": [
        {
            "name": "aNumber",
            "type": "long"
        }
        ]
    }"#;

    let reader_schema = r#"{
        "type": "record",
        "name": "ASchema",
        "fields": [
        {
            "name": "aNumber",
            "type": "int"
        }
        ]
    }"#;

    // value that was written with writer_schema
    let value = Value::Record(vec![(
            "aNumber".into(),
            Value::Long(300000000000)
    )
    ]);

    let buffer = TestBuffer(Rc::new(RefCell::new(Vec::new())), 0);

    let writer_schema = Schema::parse_str(writer_schema)?;

    let mut writer = Writer::new(&writer_schema, buffer.clone())?;

    let _ = writer.append(value);
    let _ = writer.flush();

    let reader_schema = Schema::parse_str(reader_schema)?;
    let mut reader = Reader::with_schema(&reader_schema,buffer)?;

    let read_value = reader.next();

    match read_value {
        Some(v) => {
            panic!("Reader and writer schema are incompatible, should not be able to deserialize. Deser value: {:?}", v)
        }
        None => {}
    }
    Ok(())
}

Currently, this test fails and the value of the "aNumber" field in the reader schema is a silently overflowed integer.

Not using type names in resolution

The specification's rules for schema resolution indicate that we first compare the names of named types, and if that succeeds, perform resolution recursively on that value. Since Value does not include names for named types (fixed, record, and enum), this check is not performed.

What should we do

The fix for the long->int conversion is just to remove this arm from the resolution logic we currently have. However, I think this issue is actually a little deeper.

Schema resolution as defined by the specification requires three things, a writer schema, a reader schema and a value written with the writer schema (the value is needed for resolving which union branch of the reader schema to take, if possible).

As of now, we are performing schema resolution on just the value itself. Since the value doesn't actually have schema level information like the name of named types, there are schema resolution checks (matching via unqualified name) that are impossible with the current structure of Value. Therefore, our resolution feels completely structural. However, my reading of the resolution rules from the specification indicate that the names of named types are important and are used, hence resolution is at least quasi-nominal.

Proposed solution

Reorient Value as being a flexible user-facing way of creating Avro values. Value should include the minimum amount of information to be verified against a provided schema but nothing more. That means:

• Get rid of Value::Union. In my opinion, a union is a schema-level construct describing possible types; a concrete value inhabits a specific branch, not the union itself
• Rename Value::Enum as Value::EnumVariant. Enum is the type, variant is the value.
• Values of named types (Record, Fixed, Enum) should allow an optional name. If no name is provided, resolution is essentially the same structural resolution we have now. If a name is provided, we first perform resolution by name, and if that succeeds, perform structural resolution.

To fulfill the semantic role of a value of a given schema, we should create a new type: TypedValue that wraps both a Value and a schema and verifies that the provided Value is indeed a valid value that could have been written with the schema, performing coercion steps if needed (these coercion steps are free to be a superset of the resolution steps listed in the specification). In doing so, Value should also be transformed into a new type CompleteValue that includes information from the schema its wrapped with, like named type names and union branch numbers.

Schema resolution is then a transformation from TypedValue of one schema to a TypedValue of another by performing exactly the resolution steps listed in the specification.

To discuss

Since this has overlap with the work from #353, I was thinking of rolling these two up as one eventual PR. Let me know if the above proposal makes sense.

Go Avro!

[martin-g]

The specification's rules for schema resolution indicate that

Please cite the part of the spec that you refer to. It would make it easier for others to confirm/reject your understanding of the spec.

[KitFieldhouse]

Yes of course, my apologies. Here is the excerpt from this section of the specification I had in mind:

It is an error if the two schemas do not match. To match, one of the following must hold:

both schemas are arrays whose item types match
both schemas are maps whose value types match
both schemas are enums whose (unqualified) names match
both schemas are fixed whose sizes and (unqualified) names match
both schemas are records with the same (unqualified) name
either schema is a union
both schemas have same primitive type
the writer’s schema may be promoted to the reader’s as follows:
    int is promotable to long, float, or double
    long is promotable to float or double
    float is promotable to double
    string is promotable to bytes
    bytes is promotable to string

The section then goes on to describe additional checks to perform if both schemas are records, unions, etc.

[Kriskras99]

I think the issue you found with resolving Long to Int, is a valid issue. The code uses an as cast instead of a try_from and therefore silently truncates values.
The fact that resolving is more lenient than allowed by the specification is also a valid concern, but is not something I think needs to be fixed.

I think your suggested improvement are not usable. They are all severely breaking changes and I think they are not an improvement.

If you want resolving that matches the specification, use SchemaCompatiblity to check compatiblity before resolving. CompleteValue/TypedValue are also something you can create yourself while reading values from the reader.

[Kriskras99]

@martin-g unless you disagree, I suggest we close this issue. The current resolving logic is quite lenient, but I think making it more strict will cause more harm than good. The bug with Long -> Int has been fixed, so there's nothing actionable left in this issue.

[martin-g]

• Get rid of Value::Union. In my opinion, a union is a schema-level construct describing possible types; a concrete value inhabits a specific branch, not the union itself
• Rename Value::Enum as Value::EnumVariant. Enum is the type, variant is the value.

avro-rs/avro/src/types.rs

Lines 75 to 88 in 6fc1f96

	/// An `enum` Avro value.
	///
	/// An Enum is represented by a symbol and its position in the symbols list
	/// of its corresponding schema.
	/// This allows schema-less encoding, as well as schema resolution while
	/// reading values.
	Enum(u32, String),
	/// An `union` Avro value.
	///
	/// A Union is represented by the value it holds and its position in the type list
	/// of its corresponding schema
	/// This allows schema-less encoding, as well as schema resolution while
	/// reading values.
	Union(u32, Box<Value>),

It is kind of the case already - we just have a u32 index of the variant as an extra information.

About Enum be renamed to EnumVariant - I agree it would have been a more correct name, but I also agree with @Kriskras99 that it is not worth the troubles renaming it now. The crate is still v0.x but we try to keep the API breaks as less as possible!

• Values of named types (Record, Fixed, Enum) should allow an optional name. If no name is provided, resolution is essentially the same structural resolution we have now. If a name is provided, we first perform resolution by name, and if that succeeds, perform structural resolution.

@KitFieldhouse Would that make something possible that is currently impossible/broken ?

[Kriskras99]

I realised this morning that using the name in resolving would allow the following two schemas to resolve correctly:

[{"type": "record", "name": "A", "fields": []},{"type": "record", "name": "B", "fields": []}]

[{"type": "record", "name": "B", "fields": []},{"type": "record", "name": "A", "fields": []}]

If you do this with the current resolve, than A would incorrectly resolve to B.
I think this is best solved by changing Value::resolve(self, schema: &Schema) -> Value to Value::resolve(self, reader_schema: &Schema, writer_schema: &Schema) -> Value. We can also store the name in the value, but that would cause a lot of extra string clones.

[KitFieldhouse]

Thanks for the discussion. I apologize for not responding earlier, I've been out on vacation.

@KitFieldhouse Would that make something possible that is currently impossible/broken?

What I had in mind is exactly what Kriskras demonstrates in their above post.

As for storing name in the value, as Kriskras mentions:

We can also store the name in the value, but that would cause a lot of extra string clones.

I agree that we want to avoid making a bunch of string clones. For the fork I have been working on, copying names is not too much of an issue, as names are (usually) passed via Arc references, so for values created by reading a binary stream with a schema, we can just store this reference. For user created values, the user supplied string literal only needs to live long enough for us to verify the value against a schema, at which point we can again use the reference.

I do think optionally accepting names in value makes sense. For instance, for the schema

[{"type": "record", "name": "A", "fields": []},{"type": "record", "name": "B", "fields": []}]

With the current state of Value, to specify our record as being record "B", we would need to construct our value like so

Value::Union(1,Box::new(Value::Record(vec![])));

which, in my opinion is a roundabout way of saying what we really want to say which is that we are creating a record with name "B". I think this can lead to a footgun where if a user changes the order of the union in the schema, the "name" of the record may change or we may fail to validate all together.

I think this is best solved by changing Value::resolve(self, schema: &Schema) -> Value to Value::resolve(self, reader_schema: &Schema, writer_schema: &Schema) -> Value

A few thoughts,

• Would we check Value against writer_schema first and then, if this passes, perform our resolution against
  reader_schema?

• What do we do when the specific type written in Value disagrees with what is specified in writer_schema?
  Should we reject this case, or coerce according to what makes sense? For instance, lets say we have a writer_schema of {"type": "long"}, a reader_schema of {"type": "double"} and a value of Value::Int(123)? Should we reject this since Value does not agree with writer_schema even though there is a clear coercion here? The only situation I can think of where this disagreement between writer schema and value would occur is if the provided value was constructed by the user and not read from a binary stream.

Having TypedValue represent a type-level promise that its wrapped value and schema have been checked against each other and any necessary coercions have been performed coupled with appropriately limiting the ways in which a user can create a TypedValue provides a way of eliminating the situation where we provide a Value and a writer_schema that are not mutually compatible. Then, if we write functions that take TypedValue instead of Value we can assume the fast path that no checking between writer_schema and Value is needed.

The fact that resolving is more lenient than allowed by the specification is also a valid concern, but is not something I think needs to be fixed.

In so far as this lenience doesn't cause incorrect behavior (like the example in your post) I agree with this argument to a degree. To play devil's advocate here, I see an argument that since resolution is unambiguously defined by the specification, it should be expected that implementations follow those rules and only those rules to ensure consistent behavior between implementations. I could see someone being confused on why this crate can resolve between a reader and writer schema where the Java implementation can not. Though, maybe this is fixable via a note in the documentation.

[PookieBuns]

Hi guys. I noticed another issue, specifically wrt Unions when performing schema resolution

fn incorrect_schema_resolution_with_reader() {
    use apache_avro::{Reader, Schema, Writer, types::Record};

    let schema_str = r#"
        [
            {
                "type": "record",
                "name": "foo",
                "fields": [
                    {"name": "a", "type": "string", "default": "a"},
                    {"name": "b", "type": "string", "default": "b"},
                    {"name": "c", "type": "string", "default": "c"}
                ]
            },
            {
                "type": "record",
                "name": "bar",
                "fields": [
                    {"name": "d", "type": "string"}
                ]
            }
        ]
    "#;
    let schema = apache_avro::Schema::parse_str(schema_str).unwrap();
    apache_avro::schema_compatibility::SchemaCompatibility::can_read(&schema, &schema).unwrap();
    let value = apache_avro::types::Value::Union(
        1,
        Box::new(apache_avro::types::Value::Record(vec![(
            "d".to_string(),
            apache_avro::types::Value::String("d".to_string()),
        )])),
    );
    let encoded = apache_avro::to_avro_datum(&schema, value.clone()).unwrap();
    let without_reader =
        apache_avro::from_avro_datum(&schema, &mut std::io::Cursor::new(encoded.clone()), None)
            .unwrap();
    assert_eq!(value, without_reader);
    let with_reader =
        apache_avro::from_avro_datum(&schema, &mut std::io::Cursor::new(encoded), Some(&schema))
            .unwrap();
    assert_eq!(value, with_reader);
}

In this, if you use a reader's schema, it ends up converting Bar record into Foo record because of how resolution is currently implemented.

avro-rs/avro/src/schema/union.rs

Line 87 in 48dfe5b

pub fn find_schema_with_known_schemata<S: Borrow<Schema> + Debug>(

this function currently will instantly succeed any Value in the first branch, because defaults were all provided. I haven't looked into deeply how java avoids this (maybe somewhere in https://github.com/apache/avro/blob/main/lang/java/avro/src/main/java/org/apache/avro/Resolver.java#L722) but I will take a deeper dive tmr and see if the currently proposed solution above can address it (by attaching writer schema in schema resolution)