Add tests and fixes for schema resolution bug (#9237)

# Which issue does this PR close?

- Closes #9231.

# Rationale for this change

Avro schema resolution allows a reader schema to represent “nullable”
values using a two-branch union (`["null", T]` or `[T, "null"]`) while
still reading data written with the non-union schema `T` (i.e. without
union discriminants in the encoded data).

In `arrow-avro`, resolving a non-union writer type against a reader
union (notably for array/list item schemas like `items: ["null",
"int"]`) could incorrectly treat the encoded stream as a union and
attempt to decode a union discriminant. This would misalign decoding and
could surface as `ParseError("bad varint")` for certain files (see
#9231).

# What changes are included in this PR?

- Fix schema resolution when the *writer* schema is non-union and the
*reader* schema is a union:
- Special-case two-branch unions containing `null` to be treated as
“nullable” (capturing whether `null` is first or second), and resolve
against the non-null branch.
- Improve matching for general reader unions by attempting to resolve
against each union variant, preferring a direct match, and constructing
the appropriate union resolution mapping for the selected branch.
- Ensure promotions are represented at the union-resolution level
(avoiding nested promotion resolution on the selected union child).

- Add regression coverage for the bug and the fixed behavior:
- `test_resolve_array_writer_nonunion_items_reader_nullable_items`
(schema resolution / codec)
- `test_array_decoding_writer_nonunion_items_reader_nullable_items`
(record decoding; ensures correct byte consumption and decoded values)
- `test_bad_varint_bug_nullable_array_items` (end-to-end reader
regression using a small Avro fixture)

- Add a small compressed Avro fixture under
`arrow-avro/test/data/bad-varint-bug.avro.gz` used by the regression
test.

# Are these changes tested?

Yes. This PR adds targeted unit/integration tests that reproduce the
prior failure mode and validate correct schema resolution and decoding
for nullable-union array items.

# Are there any user-facing changes?

Yes (bug fix): reading Avro files with arrays whose element type is
represented as a nullable union in the reader schema (e.g. `items:
["null", "int"]`) now succeeds instead of failing with `ParseError("bad
varint")`. No public API changes are intended.

---------

Co-authored-by: Mikhail Zabaluev <[email protected]>

Author: jecsand838

arrow-avro/src/codec.rs

@@ -1529,23 +1529,71 @@ impl<'a> Maker<'a> {
                 Ok(dt)
             }
             (writer_non_union, Schema::Union(reader_variants)) => {
-                let promo = self.find_best_promotion(
-                    writer_non_union,
-                    reader_variants.as_slice(),
-                    namespace,
-                );
-                let Some((reader_index, promotion)) = promo else {
-                    return Err(ArrowError::SchemaError(
-                        "Writer schema does not match any reader union branch".to_string(),
-                    ));
-                };
-                let mut dt = self.parse_type(reader_schema, namespace)?;
-                dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
-                    writer_to_reader: Arc::from(vec![Some((reader_index, promotion))]),
-                    writer_is_union: false,
-                    reader_is_union: true,
-                }));
-                Ok(dt)
+                if let Some((nullability, non_null_branch)) =
+                    nullable_union_variants(reader_variants)
+                {
+                    let mut dt = self.resolve_type(writer_non_union, non_null_branch, namespace)?;
+                    let non_null_idx = match nullability {
+                        Nullability::NullFirst => 1,
+                        Nullability::NullSecond => 0,
+                    };
+                    #[cfg(feature = "avro_custom_types")]
+                    Self::propagate_nullability_into_ree(&mut dt, nullability);
+                    dt.nullability = Some(nullability);
+                    let promotion = Self::coercion_from(&dt);
+                    dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
+                        writer_to_reader: Arc::from(vec![Some((non_null_idx, promotion))]),
+                        writer_is_union: false,
+                        reader_is_union: true,
+                    }));
+                    Ok(dt)
+                } else {
+                    let mut best_match: Option<(usize, AvroDataType, Promotion)> = None;
+                    for (i, variant) in reader_variants.iter().enumerate() {
+                        if let Ok(resolved_dt) =
+                            self.resolve_type(writer_non_union, variant, namespace)
+                        {
+                            let promotion = Self::coercion_from(&resolved_dt);
+                            if promotion == Promotion::Direct {
+                                best_match = Some((i, resolved_dt, promotion));
+                                break;
+                            } else if best_match.is_none() {
+                                best_match = Some((i, resolved_dt, promotion));
+                            }
+                        }
+                    }
+                    let Some((match_idx, match_dt, promotion)) = best_match else {
+                        return Err(ArrowError::SchemaError(
+                            "Writer schema does not match any reader union branch".to_string(),
+                        ));
+                    };
+                    let mut children = Vec::with_capacity(reader_variants.len());
+                    let mut match_dt = Some(match_dt);
+                    for (i, variant) in reader_variants.iter().enumerate() {
+                        if i == match_idx {
+                            if let Some(mut dt) = match_dt.take() {
+                                if matches!(dt.resolution, Some(ResolutionInfo::Promotion(_))) {
+                                    dt.resolution = None;
+                                }
+                                children.push(dt);
+                            }
+                        } else {
+                            children.push(self.parse_type(variant, namespace)?);
+                        }
+                    }
+                    let union_fields = build_union_fields(&children)?;
+                    let mut dt = AvroDataType::new(
+                        Codec::Union(children.into(), union_fields, UnionMode::Dense),
+                        Default::default(),
+                        None,
+                    );
+                    dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
+                        writer_to_reader: Arc::from(vec![Some((match_idx, promotion))]),
+                        writer_is_union: false,
+                        reader_is_union: true,
+                    }));
+                    Ok(dt)
+                }
             }
             (
                 Schema::Complex(ComplexType::Array(writer_array)),
@@ -2926,6 +2974,42 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_resolve_array_writer_nonunion_items_reader_nullable_items() {
+        let writer_schema = Schema::Complex(ComplexType::Array(Array {
+            items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
+            attributes: Attributes::default(),
+        }));
+        let reader_schema = Schema::Complex(ComplexType::Array(Array {
+            items: Box::new(mk_union(vec![
+                Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
+                Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
+            ])),
+            attributes: Attributes::default(),
+        }));
+        let mut maker = Maker::new(false, false);
+        let dt = maker
+            .make_data_type(&writer_schema, Some(&reader_schema), None)
+            .unwrap();
+        if let Codec::List(inner) = dt.codec() {
+            assert_eq!(inner.nullability(), Some(Nullability::NullFirst));
+            assert!(matches!(inner.codec(), Codec::Int32));
+            match inner.resolution.as_ref() {
+                Some(ResolutionInfo::Union(info)) => {
+                    assert!(!info.writer_is_union, "writer should be non-union");
+                    assert!(info.reader_is_union, "reader should be union");
+                    assert_eq!(
+                        info.writer_to_reader.as_ref(),
+                        &[Some((1, Promotion::Direct))]
+                    );
+                }
+                other => panic!("expected Union resolution, got {other:?}"),
+            }
+        } else {
+            panic!("expected List codec");
+        }
+    }
+
     #[test]
     fn test_resolve_fixed_success_name_and_size_match_and_alias() {
         let writer_schema = Schema::Complex(ComplexType::Fixed(Fixed {

arrow-avro/src/reader/mod.rs

@@ -9081,4 +9081,46 @@ mod test {
             "entire RecordBatch mismatch (schema, all columns, all rows)"
         );
     }
+
+    #[test]
+    fn test_bad_varint_bug_nullable_array_items() {
+        use flate2::read::GzDecoder;
+        use std::io::Read;
+        let manifest_dir = env!("CARGO_MANIFEST_DIR");
+        let gz_path = format!("{manifest_dir}/test/data/bad-varint-bug.avro.gz");
+        let gz_file = File::open(&gz_path).expect("test file should exist");
+        let mut decoder = GzDecoder::new(gz_file);
+        let mut avro_bytes = Vec::new();
+        decoder
+            .read_to_end(&mut avro_bytes)
+            .expect("should decompress");
+        let reader_arrow_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())]));
+        let reader_schema = AvroSchema::try_from(&reader_arrow_schema)
+            .expect("should convert Arrow schema to Avro");
+        let mut reader = ReaderBuilder::new()
+            .with_reader_schema(reader_schema)
+            .build(Cursor::new(avro_bytes))
+            .expect("should build reader");
+        let batch = reader
+            .next()
+            .expect("should have one batch")
+            .expect("reading should succeed without bad varint error");
+        assert_eq!(batch.num_rows(), 1);
+        let list_col = batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<ListArray>()
+            .expect("should be ListArray");
+        assert_eq!(list_col.len(), 1);
+        let values = list_col.values();
+        let int_values = values.as_primitive::<Int32Type>();
+        assert_eq!(int_values.len(), 2);
+        assert_eq!(int_values.value(0), 1);
+        assert_eq!(int_values.value(1), 2);
+    }
 }

arrow-avro/src/reader/record.rs

@@ -2897,6 +2897,54 @@ mod tests {
         assert_eq!(list_arr.value_length(0), 0);
     }
 
+    #[test]
+    fn test_array_decoding_writer_nonunion_items_reader_nullable_items() {
+        use crate::schema::Array;
+        let writer_schema = Schema::Complex(ComplexType::Array(Array {
+            items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
+            attributes: Attributes::default(),
+        }));
+        let reader_schema = Schema::Complex(ComplexType::Array(Array {
+            items: Box::new(Schema::Union(vec![
+                Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
+                Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
+            ])),
+            attributes: Attributes::default(),
+        }));
+        let dt = resolved_root_datatype(writer_schema, reader_schema, false, false);
+        if let Codec::List(inner) = dt.codec() {
+            assert_eq!(
+                inner.nullability(),
+                Some(Nullability::NullFirst),
+                "items should be nullable"
+            );
+        } else {
+            panic!("expected List codec");
+        }
+        let mut decoder = Decoder::try_new(&dt).unwrap();
+        let mut data = encode_avro_long(2);
+        data.extend(encode_avro_int(10));
+        data.extend(encode_avro_int(20));
+        data.extend(encode_avro_long(0));
+        let mut cursor = AvroCursor::new(&data);
+        decoder.decode(&mut cursor).unwrap();
+        assert_eq!(
+            cursor.position(),
+            data.len(),
+            "all bytes should be consumed"
+        );
+        let array = decoder.flush(None).unwrap();
+        let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
+        assert_eq!(list_arr.len(), 1, "one list/row");
+        assert_eq!(list_arr.value_length(0), 2, "two items in the list");
+        let values = list_arr.values().as_primitive::<Int32Type>();
+        assert_eq!(values.len(), 2);
+        assert_eq!(values.value(0), 10);
+        assert_eq!(values.value(1), 20);
+        assert!(!values.is_null(0));
+        assert!(!values.is_null(1));
+    }
+
     #[test]
     fn test_decimal_decoding_fixed256() {
         let dt = avro_from_codec(Codec::Decimal(50, Some(2), Some(32)));