Add Enum type support to arrow-avro and Minor Decimal type fix (#7852)

# Which issue does this PR close?

- Part of #4886

- Related to #6965

# Rationale for this change

The `arrow-avro` crate currently lacks support for the Avro `enum` type,
which is a standard and commonly used type in Avro schemas. This
omission prevents users from reading Avro files containing enums,
limiting the crate's utility.

This change introduces support for decoding Avro enums by mapping them
to the Arrow `DictionaryArray` type. This is a logical and efficient
representation. Implementing this feature brings the `arrow-avro` crate
closer to full Avro specification compliance and makes it more robust
for real-world use cases.

# What changes are included in this PR?

This PR introduces comprehensive support for Avro enum decoding along
with a minor Avro decimal decoding fix. The key changes are:

1.  **Schema Parsing (`codec.rs`):**
* A new `Codec::Enum(Arc<[String]>)` variant was added to represent a
parsed enum and its associated symbols.
* The `make_data_type` function now parses `ComplexType::Enum` schemas.
It also stores the original symbols as a JSON string in the `Field`'s
metadata under the key `"avro.enum.symbols"` to ensure schema fidelity
and enable lossless round-trip conversions.
* The `Codec::data_type` method was updated to map the internal
`Codec::Enum` to the corresponding Arrow
`DataType::Dictionary(Box<Int32>, Box<Utf8>)`.

2.  **Decoding Logic (`reader/record.rs`):**
* A new `Decoder::Enum(Vec<i32>, Arc<[String]>)` variant was added to
manage the state of decoding enum values.
* The `Decoder` was enhanced to create, decode, and flush `Enum` types:
        *   `try_new` creates the decoder.
        *   `decode` reads the Avro `int` index from the byte buffer.
* `flush` constructs the final `DictionaryArray<Int32Type>` using the
collected indices as keys and the stored symbols as the dictionary
values.
        *   `append_null` was extended to handle nullable enums.

3.  **Minor Decimal Type Decoding Fix (`codec.rs`)**
* A minor decimal decoding fix was implemented in `make_data_type` due
to the `(Some("decimal"), c @ Codec::Fixed(sz))` branch of `match
(t.attributes.logical_type, &mut field.codec)` not being reachable. This
issue was caught by the new decimal integration tests in
`arrow-avro/src/reader/mod.rs`.

# Are these changes tested?

*   Yes, test coverage was provided for the new `Enum` type: 
* New unit tests were added to `record.rs` to specifically validate both
non-nullable and nullable enum decoding logic.
* The existing integration test suite in `arrow-avro/src/reader/mod.rs`
was used to validate the end-to-end functionality with a new
`avro/simple_enum.avro` test case, ensuring compatibility with the
overall reader infrastructure.
 *  New tests were also included for the `Decimal` and `Fixed` types:
* This integration test suite was also extended to include tests for
`avro/simple_fixed.avro`, `avro/fixed_length_decimal.avro`,
`avro/fixed_length_decimal_legacy.avro`, `avro/int32_decimal.avro`,
`avro/int64_decimal.avro`

# Are there any user-facing changes?

N/A

Author: jecsand838

arrow-avro/src/codec.rs

@@ -203,6 +203,10 @@ pub enum Codec {
     Decimal(usize, Option<usize>, Option<usize>),
     /// Represents Avro Uuid type, a FixedSizeBinary with a length of 16
     Uuid,
+    /// Represents an Avro enum, maps to Arrow's Dictionary(Int32, Utf8) type.
+    ///
+    /// The enclosed value contains the enum's symbols.
+    Enum(Arc<[String]>),
     /// Represents Avro array type, maps to Arrow's List data type
     List(Arc<AvroDataType>),
     /// Represents Avro record type, maps to Arrow's Struct data type
@@ -253,6 +257,9 @@ impl Codec {
                 }
             }
             Self::Uuid => DataType::FixedSizeBinary(16),
+            Self::Enum(_) => {
+                DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8))
+            }
             Self::List(f) => {
                 DataType::List(Arc::new(f.field_with_name(Field::LIST_FIELD_DEFAULT_NAME)))
             }
@@ -441,7 +448,6 @@ fn make_data_type<'a>(
                         })
                     })
                     .collect::<Result<_, ArrowError>>()?;
-
                 let field = AvroDataType {
                     nullability: None,
                     codec: Codec::Struct(fields),
@@ -463,17 +469,47 @@ fn make_data_type<'a>(
                 let size = f.size.try_into().map_err(|e| {
                     ArrowError::ParseError(format!("Overflow converting size to i32: {e}"))
                 })?;
+                let md = f.attributes.field_metadata();
+                let field = match f.attributes.logical_type {
+                    Some("decimal") => {
+                        let (precision, scale, _) =
+                            parse_decimal_attributes(&f.attributes, Some(size as usize), true)?;
+                        AvroDataType {
+                            nullability: None,
+                            metadata: md,
+                            codec: Codec::Decimal(precision, Some(scale), Some(size as usize)),
+                        }
+                    }
+                    _ => AvroDataType {
+                        nullability: None,
+                        metadata: md,
+                        codec: Codec::Fixed(size),
+                    },
+                };
+                resolver.register(f.name, namespace, field.clone());
+                Ok(field)
+            }
+            ComplexType::Enum(e) => {
+                let namespace = e.namespace.or(namespace);
+                let symbols = e
+                    .symbols
+                    .iter()
+                    .map(|s| s.to_string())
+                    .collect::<Arc<[String]>>();
+
+                let mut metadata = e.attributes.field_metadata();
+                let symbols_json = serde_json::to_string(&e.symbols).map_err(|e| {
+                    ArrowError::ParseError(format!("Failed to serialize enum symbols: {e}"))
+                })?;
+                metadata.insert("avro.enum.symbols".to_string(), symbols_json);
                 let field = AvroDataType {
                     nullability: None,
-                    metadata: f.attributes.field_metadata(),
-                    codec: Codec::Fixed(size),
+                    metadata,
+                    codec: Codec::Enum(symbols),
                 };
-                resolver.register(f.name, namespace, field.clone());
+                resolver.register(e.name, namespace, field.clone());
                 Ok(field)
             }
-            ComplexType::Enum(e) => Err(ArrowError::NotYetImplemented(format!(
-                "Enum of {e:?} not currently supported"
-            ))),
             ComplexType::Map(m) => {
                 let val = make_data_type(&m.values, namespace, resolver, use_utf8view)?;
                 Ok(AvroDataType {
@@ -493,27 +529,10 @@ fn make_data_type<'a>(
 
             // https://avro.apache.org/docs/1.11.1/specification/#logical-types
             match (t.attributes.logical_type, &mut field.codec) {
-                (Some("decimal"), c) => match *c {
-                    Codec::Fixed(sz_val) => {
-                        let (prec, sc, size_opt) =
-                            parse_decimal_attributes(&t.attributes, Some(sz_val as usize), true)?;
-                        let final_sz = if let Some(sz_actual) = size_opt {
-                            sz_actual
-                        } else {
-                            sz_val as usize
-                        };
-                        *c = Codec::Decimal(prec, Some(sc), Some(final_sz));
-                    }
-                    Codec::Binary => {
-                        let (prec, sc, _) = parse_decimal_attributes(&t.attributes, None, false)?;
-                        *c = Codec::Decimal(prec, Some(sc), None);
-                    }
-                    _ => {
-                        return Err(ArrowError::SchemaError(format!(
-                            "Decimal logical type can only be backed by Fixed or Bytes, found {c:?}"
-                        )))
-                    }
-                },
+                (Some("decimal"), c @ Codec::Binary) => {
+                    let (prec, sc, _) = parse_decimal_attributes(&t.attributes, None, false)?;
+                    *c = Codec::Decimal(prec, Some(sc), None);
+                }
                 (Some("date"), c @ Codec::Int32) => *c = Codec::Date32,
                 (Some("time-millis"), c @ Codec::Int32) => *c = Codec::TimeMillis,
                 (Some("time-micros"), c @ Codec::Int64) => *c = Codec::TimeMicros,

arrow-avro/src/reader/mod.rs

@@ -121,8 +121,9 @@ mod test {
     use crate::reader::record::RecordDecoder;
     use crate::reader::{read_blocks, read_header};
     use crate::test_util::arrow_test_data;
+    use arrow_array::types::Int32Type;
     use arrow_array::*;
-    use arrow_schema::{DataType, Field};
+    use arrow_schema::{DataType, Field, Schema};
     use std::collections::HashMap;
     use std::fs::File;
     use std::io::BufReader;
@@ -150,20 +151,26 @@ mod test {
         for result in read_blocks(reader) {
             let block = result.unwrap();
             assert_eq!(block.sync, header.sync());
-            if let Some(c) = compression {
-                let decompressed = c.decompress(&block.data).unwrap();
 
+            let mut decode_data = |data: &[u8]| {
                 let mut offset = 0;
                 let mut remaining = block.count;
                 while remaining > 0 {
-                    let to_read = remaining.max(batch_size);
-                    offset += decoder
-                        .decode(&decompressed[offset..], block.count)
-                        .unwrap();
-
+                    let to_read = remaining.min(batch_size);
+                    if to_read == 0 {
+                        break;
+                    }
+                    offset += decoder.decode(&data[offset..], to_read).unwrap();
                     remaining -= to_read;
                 }
-                assert_eq!(offset, decompressed.len());
+                assert_eq!(offset, data.len());
+            };
+
+            if let Some(c) = compression {
+                let decompressed = c.decompress(&block.data).unwrap();
+                decode_data(&decompressed);
+            } else {
+                decode_data(&block.data);
             }
         }
         decoder.flush().unwrap()
@@ -308,4 +315,129 @@ mod test {
             assert_eq!(read_file(&file, 3), expected);
         }
     }
+
+    #[test]
+    fn test_decimal() {
+        let files = [
+            ("avro/fixed_length_decimal.avro", 25, 2),
+            ("avro/fixed_length_decimal_legacy.avro", 13, 2),
+            ("avro/int32_decimal.avro", 4, 2),
+            ("avro/int64_decimal.avro", 10, 2),
+        ];
+        let decimal_values: Vec<i128> = (1..=24).map(|n| n as i128 * 100).collect();
+        for (file, precision, scale) in files {
+            let file_path = arrow_test_data(file);
+            let actual_batch = read_file(&file_path, 8);
+            let expected_array = Decimal128Array::from_iter_values(decimal_values.clone())
+                .with_precision_and_scale(precision, scale)
+                .unwrap();
+            let mut meta = HashMap::new();
+            meta.insert("precision".to_string(), precision.to_string());
+            meta.insert("scale".to_string(), scale.to_string());
+            let field_with_meta = Field::new("value", DataType::Decimal128(precision, scale), true)
+                .with_metadata(meta);
+            let expected_schema = Arc::new(Schema::new(vec![field_with_meta]));
+            let expected_batch =
+                RecordBatch::try_new(expected_schema.clone(), vec![Arc::new(expected_array)])
+                    .expect("Failed to build expected RecordBatch");
+            assert_eq!(
+                actual_batch, expected_batch,
+                "Decoded RecordBatch does not match the expected Decimal128 data for file {file}"
+            );
+            let actual_batch_small = read_file(&file_path, 3);
+            assert_eq!(
+                actual_batch_small,
+                expected_batch,
+                "Decoded RecordBatch does not match the expected Decimal128 data for file {file} with batch size 3"
+            );
+        }
+    }
+
+    #[test]
+    fn test_simple() {
+        let tests = [
+            ("avro/simple_enum.avro", 4, build_expected_enum(), 2),
+            ("avro/simple_fixed.avro", 2, build_expected_fixed(), 1),
+        ];
+
+        fn build_expected_enum() -> RecordBatch {
+            // Build the DictionaryArrays for f1, f2, f3
+            let keys_f1 = Int32Array::from(vec![0, 1, 2, 3]);
+            let vals_f1 = StringArray::from(vec!["a", "b", "c", "d"]);
+            let f1_dict =
+                DictionaryArray::<Int32Type>::try_new(keys_f1, Arc::new(vals_f1)).unwrap();
+            let keys_f2 = Int32Array::from(vec![2, 3, 0, 1]);
+            let vals_f2 = StringArray::from(vec!["e", "f", "g", "h"]);
+            let f2_dict =
+                DictionaryArray::<Int32Type>::try_new(keys_f2, Arc::new(vals_f2)).unwrap();
+            let keys_f3 = Int32Array::from(vec![Some(1), Some(2), None, Some(0)]);
+            let vals_f3 = StringArray::from(vec!["i", "j", "k"]);
+            let f3_dict =
+                DictionaryArray::<Int32Type>::try_new(keys_f3, Arc::new(vals_f3)).unwrap();
+            let dict_type =
+                DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
+            let mut md_f1 = HashMap::new();
+            md_f1.insert(
+                "avro.enum.symbols".to_string(),
+                r#"["a","b","c","d"]"#.to_string(),
+            );
+            let f1_field = Field::new("f1", dict_type.clone(), false).with_metadata(md_f1);
+            let mut md_f2 = HashMap::new();
+            md_f2.insert(
+                "avro.enum.symbols".to_string(),
+                r#"["e","f","g","h"]"#.to_string(),
+            );
+            let f2_field = Field::new("f2", dict_type.clone(), false).with_metadata(md_f2);
+            let mut md_f3 = HashMap::new();
+            md_f3.insert(
+                "avro.enum.symbols".to_string(),
+                r#"["i","j","k"]"#.to_string(),
+            );
+            let f3_field = Field::new("f3", dict_type.clone(), true).with_metadata(md_f3);
+            let expected_schema = Arc::new(Schema::new(vec![f1_field, f2_field, f3_field]));
+            RecordBatch::try_new(
+                expected_schema,
+                vec![
+                    Arc::new(f1_dict) as Arc<dyn Array>,
+                    Arc::new(f2_dict) as Arc<dyn Array>,
+                    Arc::new(f3_dict) as Arc<dyn Array>,
+                ],
+            )
+            .unwrap()
+        }
+
+        fn build_expected_fixed() -> RecordBatch {
+            let f1 =
+                FixedSizeBinaryArray::try_from_iter(vec![b"abcde", b"12345"].into_iter()).unwrap();
+            let f2 =
+                FixedSizeBinaryArray::try_from_iter(vec![b"fghijklmno", b"1234567890"].into_iter())
+                    .unwrap();
+            let f3 = FixedSizeBinaryArray::try_from_sparse_iter_with_size(
+                vec![Some(b"ABCDEF" as &[u8]), None].into_iter(),
+                6,
+            )
+            .unwrap();
+            let expected_schema = Arc::new(Schema::new(vec![
+                Field::new("f1", DataType::FixedSizeBinary(5), false),
+                Field::new("f2", DataType::FixedSizeBinary(10), false),
+                Field::new("f3", DataType::FixedSizeBinary(6), true),
+            ]));
+            RecordBatch::try_new(
+                expected_schema,
+                vec![
+                    Arc::new(f1) as Arc<dyn Array>,
+                    Arc::new(f2) as Arc<dyn Array>,
+                    Arc::new(f3) as Arc<dyn Array>,
+                ],
+            )
+            .unwrap()
+        }
+        for (file_name, batch_size, expected, alt_batch_size) in tests {
+            let file = arrow_test_data(file_name);
+            let actual = read_file(&file, batch_size);
+            assert_eq!(actual, expected);
+            let actual2 = read_file(&file, alt_batch_size);
+            assert_eq!(actual2, expected);
+        }
+    }
 }