`ParseError("bad varint")` on reading array field with nullable elements in the reader schema

[mzabaluev]

I get a ParseError("bad varint") on this test (added to ./arrow-avro/src/reader/mod.rs):

    fn get_int_array_schema() -> AvroSchema {
        let schema = Schema::new(vec![Field::new(
            "int_array",
            DataType::List(Arc::new(Field::new("element", DataType::Int32, true))),
            true,
        )])
        .with_metadata(HashMap::from([("avro.name".into(), "table".into())]));
        AvroSchema::try_from(&schema).unwrap()
    }

    #[test]
    fn test_bad_varint_bug() {
        let path = arrow_test_data("avro/bad-varint-bug.avro");

        let schema = get_int_array_schema();

        let batch = read_alltypes_with_reader_schema(&path, schema);

        let int_list_col = batch.column(0).as_list::<i32>();
        let first_list = int_list_col.value(0);
        let expected_result = Arc::new(Int32Array::from_iter_values(vec![1i32, 2])) as _;
        assert_eq!(first_list, expected_result)
    }

The Avro file, readable by Spark: bad-varint-bug.avro.gz

Originally posted by @mzabaluev in #8930 (comment)

[mzabaluev]

If the nullability of the array field ("element") is changed to false, the test passes.

The writer schema prints out as follows:

Complex(
    Record(
        Record {
            name: "table",
            namespace: None,
            doc: None,
            aliases: [],
            fields: [
                Field {
                    name: "int_array",
                    doc: None,
                    type: Complex(
                        Array(
                            Array {
                                items: TypeName(
                                    Primitive(
                                        Int,
                                    ),
                                ),
                                attributes: Attributes {
                                    logical_type: None,
                                    additional: {
                                        "element-id": Number(2),
                                    },
                                },
                            },
                        ),
                    ),
                    default: None,
                    aliases: [],
                },
            ],
            attributes: Attributes {
                logical_type: None,
                additional: {},
            },
        },
    ),
)

Should the nullable reader schema extend the non-nullable writer schema?

[jecsand838]

@mzabaluev

Thanks for the clear reproducer and sample file.

This looks like a schema-resolution bug in arrow-avro. Per the Avro spec, when the reader schema is a union but the writer is not, the reader must select the first matching union branch and then recursively resolve against it (i.e., decode using the writer’s encoding with no union tag consumed).

In this case, the writer has array<int>, and the reader requests array<union<null,int>>, which should be compatible.

The fix for this should be implementable in the (writer_non_union, reader_union) resolver arm in codec.rs. I'll work on getting a PR up for this tonight.

Should the nullable reader schema extend the non-nullable writer schema?

Yes -- according to the Avro spec, this is explicitly supported:

• If reader is a union but writer is not, the reader must pick the first union branch that matches the writer schema and then that branch is recursively resolved against the writer schema.
• And because arrays are resolved recursively on their item schemas, this applies to array elements too.

So:

• writer: array
• reader: array<["null","int"]>

should be compatible, and Spark successfully reading the file is consistent with this expected behavior.

[jecsand838]

take

[mzabaluev]

The inner element type gets resolved to this codec instance, which seems correct for decoding non-nullable elements:

AvroDataType {
    nullability: Some(
        NullFirst,
    ),
    metadata: {},
    codec: Int32,
    resolution: Some(
        Union(
            ResolvedUnion {
                writer_to_reader: [
                    Some(
                        (
                            1,
                            Direct,
                        ),
                    ),
                ],
                writer_is_union: false,
                reader_is_union: true,
            },
        ),
    ),
}

But when the array type codec is constructed, the resolution information on the element type is lost:

AvroDataType {
    nullability: Some(
        NullFirst,
    ),
    metadata: {},
    codec: List(
        AvroDataType {
            nullability: Some(
                NullFirst,
            ),
            metadata: {},
            codec: Int32,
            resolution: None,
        },
    ),
    resolution: Some(
        Union(
            ResolvedUnion {
                writer_to_reader: [
                    Some(
                        (
                            1,
                            Direct,
                        ),
                    ),
                ],
                writer_is_union: false,
                reader_is_union: true,
            },
        ),
    ),
}

[jecsand838]

@mzabaluev Just pushed a fix for this. Let me know if it looks good to you.