Regression Testing, Bug Fixes, and Public API Tightening for arrow-avro (#8492)

# Which issue does this PR close?

- **Related to**: #4886 (“Add Avro Support”)

# Rationale for this change

**NOTE:** This PR contains over **2300 lines of test code**. The actual
production code diff is **less than 800 LOC**.

Before we publish `arrow-avro`, we want to "minimize its public API
surface" and ship a well‑tested, spec‑compliant implementation. In the
process of adding intensive regression tests and canonical‑form checks,
we found several correctness gaps around alias handling, union
resolution, Unicode/name validation, list child nullability, “null”
string handling, and a mis-wired `Writer` capacity setting. This PR
tightens the API and fixes those issues to align with the Avro spec
(aliases and defaults, union resolution, names and Unicode, etc.).

# What changes are included in this PR?

**Public API tightening**
- Restrict visibility of numerous schema/codec types and functions
within `arrow-avro` so only intended entry points are public ahead of
making the crate public.

**Bug fixes discovered via regression testing (All fixed)**
1. **Alias bugs (aliases without defaults / restrictive namespaces)**  
- Enforce spec‑compliant alias resolution: aliases may be
fully‑qualified or relative to the reader’s namespace, and alias‑based
rewrites still require reader defaults when the writer field is absent.
This follows Avro’s alias rules and record‑field default behavior.
2. **Special‑case union resolution (writer not a union, reader is)**  
- When the writer schema is **not** a union but the reader is, we no
longer attempt to decode a union `type_id`; per spec, the reader must
pick the first union branch that matches the writer’s schema.
3. **Valid Avro Unicode characters & name rules in Schema**  
- Distinguish between *Unicode strings* (which may contain any valid
UTF‑8) and *identifiers* (names/enum symbols) which must match
`[A-Za-z_][A-Za-z0-9_]*`. Tests were added to accept valid Unicode
string content while enforcing the ASCII identifier regex.
4. **Nullable `ListArray` child item bug**  
- Correct mapping of Avro array item nullability to Arrow `ListArray`’s
inner `"item"` field. (By convention the inner field is named `"item"`
and nullability is explicit.) This aligns with Arrow’s builder/typing
docs.
5. **“null” string vs. hard `null`**  
- Fix default/value handling to differentiate JSON `null` from the
string literal `"null"` per the Avro defaults table.
6. **`Writer` capacity knob wired up**  
- Plumb the provided capacity through the writer implementation so
preallocation/knobbed capacity is respected. (See changes under
`arrow-avro/src/writer/mod.rs`.)

# Are these changes tested?

Yes. This PR adds substantial regression coverage:
- Canonical‑form checks for schemas.
- Alias/namespace + default‑value resolution cases.
- Reader‑union vs. writer‑non‑union decoding paths.
- Unicode content vs. identifier name rules.
- `ListArray` inner field nullability behavior.
- Round‑trips exercising the `Writer` with the capacity knob set.
  
A new, comprehensive Avro fixture (`test/data/comprehensive_e2e.avro`)
is included to drive end‑to‑end scenarios and edge cases,.

# Are there any user-facing changes?

N/A

Author: jecsand838

arrow-avro/src/codec.rs

@@ -91,7 +91,7 @@ impl Header {
         }
     }
 
-    /// Returns the [`Schema`] if any
+    /// Returns the `Schema` if any
     pub(crate) fn schema(&self) -> Result<Option<Schema<'_>>, ArrowError> {
         self.get(SCHEMA_METADATA_KEY)
             .map(|x| {

arrow-avro/src/reader/header.rs

@@ -40,6 +40,16 @@ use uuid::Uuid;
 
 const DEFAULT_CAPACITY: usize = 1024;
 
+/// Runtime plan for decoding reader-side `["null", T]` types.
+#[derive(Clone, Copy, Debug)]
+enum NullablePlan {
+    /// Writer actually wrote a union (branch tag present).
+    ReadTag,
+    /// Writer wrote a single (non-union) value resolved to the non-null branch
+    /// of the reader union; do NOT read a branch tag, but apply any promotion.
+    FromSingle { promotion: Promotion },
+}
+
 /// Macro to decode a decimal payload for a given width and integer type.
 macro_rules! decode_decimal {
     ($size:expr, $buf:expr, $builder:expr, $N:expr, $Int:ty) => {{
@@ -267,7 +277,7 @@ enum Decoder {
     Decimal128(usize, Option<usize>, Option<usize>, Decimal128Builder),
     Decimal256(usize, Option<usize>, Option<usize>, Decimal256Builder),
     Union(UnionDecoder),
-    Nullable(Nullability, NullBufferBuilder, Box<Decoder>),
+    Nullable(Nullability, NullBufferBuilder, Box<Decoder>, NullablePlan),
 }
 
 impl Decoder {
@@ -508,11 +518,23 @@ impl Decoder {
             }
         };
         Ok(match data_type.nullability() {
-            Some(nullability) => Self::Nullable(
-                nullability,
-                NullBufferBuilder::new(DEFAULT_CAPACITY),
-                Box::new(decoder),
-            ),
+            Some(nullability) => {
+                // Default to reading a union branch tag unless the resolution proves otherwise.
+                let mut plan = NullablePlan::ReadTag;
+                if let Some(ResolutionInfo::Union(info)) = data_type.resolution.as_ref() {
+                    if !info.writer_is_union && info.reader_is_union {
+                        if let Some(Some((_reader_idx, promo))) = info.writer_to_reader.first() {
+                            plan = NullablePlan::FromSingle { promotion: *promo };
+                        }
+                    }
+                }
+                Self::Nullable(
+                    nullability,
+                    NullBufferBuilder::new(DEFAULT_CAPACITY),
+                    Box::new(decoder),
+                    plan,
+                )
+            }
             None => decoder,
         })
     }
@@ -571,7 +593,7 @@ impl Decoder {
             Self::Enum(indices, _, _) => indices.push(0),
             Self::Duration(builder) => builder.append_null(),
             Self::Union(u) => u.append_null()?,
-            Self::Nullable(_, null_buffer, inner) => {
+            Self::Nullable(_, null_buffer, inner, _) => {
                 null_buffer.append(false);
                 inner.append_null();
             }
@@ -582,7 +604,7 @@ impl Decoder {
     /// Append a single default literal into the decoder's buffers
     fn append_default(&mut self, lit: &AvroLiteral) -> Result<(), ArrowError> {
         match self {
-            Self::Nullable(_, nb, inner) => {
+            Self::Nullable(_, nb, inner, _) => {
                 if matches!(lit, AvroLiteral::Null) {
                     nb.append(false);
                     inner.append_null()
@@ -939,19 +961,27 @@ impl Decoder {
                 builder.append_value(IntervalMonthDayNano::new(months as i32, days as i32, nanos));
             }
             Self::Union(u) => u.decode(buf)?,
-            Self::Nullable(order, nb, encoding) => {
-                let branch = buf.read_vlq()?;
-                let is_not_null = match *order {
-                    Nullability::NullFirst => branch != 0,
-                    Nullability::NullSecond => branch == 0,
-                };
-                if is_not_null {
-                    // It is important to decode before appending to null buffer in case of decode error
-                    encoding.decode(buf)?;
-                } else {
-                    encoding.append_null();
+            Self::Nullable(order, nb, encoding, plan) => {
+                match *plan {
+                    NullablePlan::FromSingle { promotion } => {
+                        encoding.decode_with_promotion(buf, promotion)?;
+                        nb.append(true);
+                    }
+                    NullablePlan::ReadTag => {
+                        let branch = buf.read_vlq()?;
+                        let is_not_null = match *order {
+                            Nullability::NullFirst => branch != 0,
+                            Nullability::NullSecond => branch == 0,
+                        };
+                        if is_not_null {
+                            // It is important to decode before appending to null buffer in case of decode error
+                            encoding.decode(buf)?;
+                        } else {
+                            encoding.append_null();
+                        }
+                        nb.append(is_not_null);
+                    }
                 }
-                nb.append(is_not_null);
             }
         }
         Ok(())
@@ -1018,7 +1048,7 @@ impl Decoder {
     /// Flush decoded records to an [`ArrayRef`]
     fn flush(&mut self, nulls: Option<NullBuffer>) -> Result<ArrayRef, ArrowError> {
         Ok(match self {
-            Self::Nullable(_, n, e) => e.flush(n.finish())?,
+            Self::Nullable(_, n, e, _) => e.flush(n.finish())?,
             Self::Null(size) => Arc::new(NullArray::new(std::mem::replace(size, 0))),
             Self::Boolean(b) => Arc::new(BooleanArray::new(b.finish(), nulls)),
             Self::Int32(values) => Arc::new(flush_primitive::<Int32Type>(values, nulls)),
@@ -2742,6 +2772,7 @@ mod tests {
             Nullability::NullSecond,
             NullBufferBuilder::new(DEFAULT_CAPACITY),
             Box::new(inner),
+            NullablePlan::ReadTag,
         );
         let mut data = Vec::new();
         data.extend_from_slice(&encode_avro_int(0));
@@ -2784,6 +2815,7 @@ mod tests {
             Nullability::NullSecond,
             NullBufferBuilder::new(DEFAULT_CAPACITY),
             Box::new(inner),
+            NullablePlan::ReadTag,
         );
         let row1 = [
             0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
@@ -3663,6 +3695,7 @@ mod tests {
             Nullability::NullFirst,
             NullBufferBuilder::new(DEFAULT_CAPACITY),
             Box::new(inner),
+            NullablePlan::ReadTag,
         );
         dec.append_default(&AvroLiteral::Null).unwrap();
         dec.append_default(&AvroLiteral::Int(11)).unwrap();
@@ -3916,6 +3949,7 @@ mod tests {
             Nullability::NullSecond,
             NullBufferBuilder::new(DEFAULT_CAPACITY),
             Box::new(Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY))),
+            NullablePlan::ReadTag,
         );
         let enc_b = Decoder::Nullable(
             Nullability::NullSecond,
@@ -3924,6 +3958,7 @@ mod tests {
                 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
                 Vec::with_capacity(DEFAULT_CAPACITY),
             )),
+            NullablePlan::ReadTag,
         );
         encoders.push(enc_a);
         encoders.push(enc_b);