Add support for Duration type and minor fixes for UUID in arrow-avro:

- Fixed `Uuid` support, now represented as `Utf8` in Arrow and added testing logic.
- Added `Duration` support, mapped to Arrow's `IntervalMonthDayNano`, with schema handling, decoding, and integration tests.
- Updated `Cargo.toml` to include the `uuid` crate as a dev dependency for UUID checking.
- Added integration tests with the new `duration_uuid.avro` test file.

Author: jecsand838

arrow-avro/Cargo.toml

@@ -56,6 +56,7 @@ rand = { version = "0.9", default-features = false, features = ["std", "std_rng"
 criterion = { version = "0.6.0", default-features = false }
 tempfile = "3.3"
 arrow = { workspace = true }
+uuid = { version = "1.17.0", features = ["v4"] }
 
 [[bench]]
 name = "avro_reader"

arrow-avro/src/codec.rs

@@ -256,7 +256,7 @@ impl Codec {
                     Decimal128(p, s)
                 }
             }
-            Self::Uuid => DataType::FixedSizeBinary(16),
+            Self::Uuid => DataType::Utf8,
             Self::Enum(_) => {
                 DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8))
             }
@@ -480,6 +480,18 @@ fn make_data_type<'a>(
                             codec: Codec::Decimal(precision, Some(scale), Some(size as usize)),
                         }
                     }
+                    Some("duration") => {
+                        if size != 12 {
+                            return Err(ArrowError::ParseError(format!(
+                                "Invalid fixed size for Duration: {size}, must be 12"
+                            )));
+                        };
+                        AvroDataType {
+                            nullability: None,
+                            metadata: md,
+                            codec: Codec::Interval,
+                        }
+                    }
                     _ => AvroDataType {
                         nullability: None,
                         metadata: md,
@@ -544,7 +556,6 @@ fn make_data_type<'a>(
                 (Some("local-timestamp-micros"), c @ Codec::Int64) => {
                     *c = Codec::TimestampMicros(false)
                 }
-                (Some("duration"), c @ Codec::Fixed(12)) => *c = Codec::Interval,
                 (Some("uuid"), c @ Codec::Utf8) => *c = Codec::Uuid,
                 (Some(logical), _) => {
                     // Insert unrecognized logical type into metadata map

arrow-avro/src/reader/mod.rs

@@ -121,13 +121,14 @@ 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::types::{Int32Type, IntervalMonthDayNanoType};
     use arrow_array::*;
-    use arrow_schema::{DataType, Field, Schema};
+    use arrow_schema::{DataType, Field, IntervalUnit, Schema};
     use std::collections::HashMap;
     use std::fs::File;
     use std::io::BufReader;
     use std::sync::Arc;
+    use uuid::Uuid;
 
     fn read_file(file: &str, batch_size: usize) -> RecordBatch {
         read_file_with_options(file, batch_size, &crate::ReadOptions::default())
@@ -440,4 +441,45 @@ mod test {
             assert_eq!(actual2, expected);
         }
     }
+
+    #[test]
+    fn test_duration_uuid() {
+        let batch = read_file("test/data/duration_uuid.avro", 4);
+        let schema = batch.schema();
+        let fields = schema.fields();
+        assert_eq!(fields.len(), 2);
+        assert_eq!(fields[0].name(), "duration_field");
+        assert_eq!(
+            fields[0].data_type(),
+            &DataType::Interval(IntervalUnit::MonthDayNano)
+        );
+        assert_eq!(fields[1].name(), "uuid_field");
+        assert_eq!(fields[1].data_type(), &DataType::Utf8);
+        assert_eq!(batch.num_rows(), 4);
+        assert_eq!(batch.num_columns(), 2);
+        let duration_array = batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<IntervalMonthDayNanoArray>()
+            .unwrap();
+        let expected_duration_array: IntervalMonthDayNanoArray = [
+            Some(IntervalMonthDayNanoType::make_value(1, 15, 500_000_000)),
+            Some(IntervalMonthDayNanoType::make_value(0, 5, 2_500_000_000)),
+            Some(IntervalMonthDayNanoType::make_value(2, 0, 0)),
+            Some(IntervalMonthDayNanoType::make_value(12, 31, 999_000_000)),
+        ]
+        .iter()
+        .copied()
+        .collect();
+        assert_eq!(&expected_duration_array, duration_array);
+        let uuid_array = batch
+            .column(1)
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap();
+        for i in 0..uuid_array.len() {
+            assert!(uuid_array.is_valid(i));
+            assert!(Uuid::parse_str(uuid_array.value(i)).is_ok());
+        }
+    }
 }

arrow-avro/src/reader/record.rs

@@ -21,13 +21,16 @@ use crate::reader::cursor::AvroCursor;
 use crate::reader::header::Header;
 use crate::reader::ReadOptions;
 use crate::schema::*;
-use arrow_array::builder::{Decimal128Builder, Decimal256Builder};
+use arrow_array::builder::{
+    ArrayBuilder, Decimal128Builder, Decimal256Builder, IntervalMonthDayNanoBuilder,
+    PrimitiveBuilder,
+};
 use arrow_array::types::*;
 use arrow_array::*;
 use arrow_buffer::*;
 use arrow_schema::{
-    ArrowError, DataType, Field as ArrowField, FieldRef, Fields, Schema as ArrowSchema, SchemaRef,
-    DECIMAL128_MAX_PRECISION, DECIMAL256_MAX_PRECISION,
+    ArrowError, DataType, Field as ArrowField, FieldRef, Fields, IntervalUnit,
+    Schema as ArrowSchema, SchemaRef, DECIMAL128_MAX_PRECISION, DECIMAL256_MAX_PRECISION,
 };
 use std::cmp::Ordering;
 use std::collections::HashMap;
@@ -128,15 +131,15 @@ enum Decoder {
     ),
     Fixed(i32, Vec<u8>),
     Enum(Vec<i32>, Arc<[String]>),
+    Duration(IntervalMonthDayNanoBuilder),
+    Uuid(OffsetBufferBuilder<i32>, Vec<u8>),
     Decimal128(usize, Option<usize>, Option<usize>, Decimal128Builder),
     Decimal256(usize, Option<usize>, Option<usize>, Decimal256Builder),
     Nullable(Nullability, NullBufferBuilder, Box<Decoder>),
 }
 
 impl Decoder {
     fn try_new(data_type: &AvroDataType) -> Result<Self, ArrowError> {
-        let nyi = |s: &str| Err(ArrowError::NotYetImplemented(s.to_string()));
-
         let decoder = match data_type.codec() {
             Codec::Null => Self::Null(0),
             Codec::Boolean => Self::Boolean(BooleanBufferBuilder::new(DEFAULT_CAPACITY)),
@@ -205,7 +208,7 @@ impl Decoder {
                     }
                 }
             }
-            Codec::Interval => return nyi("decoding interval"),
+            Codec::Interval => Self::Duration(IntervalMonthDayNanoBuilder::new()),
             Codec::List(item) => {
                 let decoder = Self::try_new(item)?;
                 Self::Array(
@@ -246,7 +249,10 @@ impl Decoder {
                     Box::new(val_dec),
                 )
             }
-            Codec::Uuid => Self::Fixed(16, Vec::with_capacity(DEFAULT_CAPACITY)),
+            Codec::Uuid => Self::Uuid(
+                OffsetBufferBuilder::new(DEFAULT_CAPACITY),
+                Vec::with_capacity(DEFAULT_CAPACITY),
+            ),
         };
         Ok(match data_type.nullability() {
             Some(nullability) => Self::Nullable(
@@ -270,7 +276,10 @@ impl Decoder {
             | Self::TimestampMicros(_, v) => v.push(0),
             Self::Float32(v) => v.push(0.),
             Self::Float64(v) => v.push(0.),
-            Self::Binary(offsets, _) | Self::String(offsets, _) | Self::StringView(offsets, _) => {
+            Self::Binary(offsets, _)
+            | Self::String(offsets, _)
+            | Self::StringView(offsets, _)
+            | Self::Uuid(offsets, _) => {
                 offsets.push_length(0);
             }
             Self::Array(_, offsets, e) => {
@@ -287,6 +296,7 @@ impl Decoder {
             Self::Decimal128(_, _, _, builder) => builder.append_value(0),
             Self::Decimal256(_, _, _, builder) => builder.append_value(i256::ZERO),
             Self::Enum(indices, _) => indices.push(0),
+            Self::Duration(builder) => builder.append_null(),
             Self::Nullable(_, _, _) => unreachable!("Nulls cannot be nested"),
         }
     }
@@ -307,7 +317,8 @@ impl Decoder {
             Self::Float64(values) => values.push(buf.get_double()?),
             Self::Binary(offsets, values)
             | Self::String(offsets, values)
-            | Self::StringView(offsets, values) => {
+            | Self::StringView(offsets, values)
+            | Self::Uuid(offsets, values) => {
                 let data = buf.get_bytes()?;
                 offsets.push_length(data.len());
                 values.extend_from_slice(data);
@@ -357,6 +368,14 @@ impl Decoder {
             Self::Enum(indices, _) => {
                 indices.push(buf.get_int()?);
             }
+            Self::Duration(builder) => {
+                let b = buf.get_fixed(12)?;
+                let months = u32::from_le_bytes(b[0..4].try_into().unwrap());
+                let days = u32::from_le_bytes(b[4..8].try_into().unwrap());
+                let millis = u32::from_le_bytes(b[8..12].try_into().unwrap());
+                let nanos = (millis as i64) * 1_000_000;
+                builder.append_value(IntervalMonthDayNano::new(months as i32, days as i32, nanos));
+            }
             Self::Nullable(nullability, nulls, e) => {
                 let is_valid = buf.get_bool()? == matches!(nullability, Nullability::NullFirst);
                 nulls.append(is_valid);
@@ -399,7 +418,7 @@ impl Decoder {
                 let values = flush_values(values).into();
                 Arc::new(BinaryArray::new(offsets, values, nulls))
             }
-            Self::String(offsets, values) => {
+            Self::String(offsets, values) | Self::Uuid(offsets, values) => {
                 let offsets = flush_offsets(offsets);
                 let values = flush_values(values).into();
                 Arc::new(StringArray::new(offsets, values, nulls))
@@ -417,7 +436,6 @@ impl Decoder {
                         }
                     })
                     .collect();
-
                 Arc::new(StringViewArray::from(values))
             }
             Self::Array(field, offsets, values) => {
@@ -472,17 +490,15 @@ impl Decoder {
                 Arc::new(arr)
             }
             Self::Decimal128(precision, scale, _, builder) => {
-                let mut b = std::mem::take(builder);
-                let (_, vals, _) = b.finish().into_parts();
+                let (_, vals, _) = builder.finish().into_parts();
                 let scl = scale.unwrap_or(0);
                 let dec = Decimal128Array::new(vals, nulls)
                     .with_precision_and_scale(*precision as u8, scl as i8)
                     .map_err(|e| ArrowError::ParseError(e.to_string()))?;
                 Arc::new(dec)
             }
             Self::Decimal256(precision, scale, _, builder) => {
-                let mut b = std::mem::take(builder);
-                let (_, vals, _) = b.finish().into_parts();
+                let (_, vals, _) = builder.finish().into_parts();
                 let scl = scale.unwrap_or(0);
                 let dec = Decimal256Array::new(vals, nulls)
                     .with_precision_and_scale(*precision as u8, scl as i8)
@@ -496,6 +512,12 @@ impl Decoder {
                 ));
                 Arc::new(DictionaryArray::try_new(keys, values)?)
             }
+            Self::Duration(builder) => {
+                let (_, vals, _) = builder.finish().into_parts();
+                let vals = IntervalMonthDayNanoArray::try_new(vals, nulls)
+                    .map_err(|e| ArrowError::ParseError(e.to_string()))?;
+                Arc::new(vals)
+            }
         })
     }
 }
@@ -744,16 +766,16 @@ mod tests {
     fn test_uuid_decoding() {
         let avro_type = avro_from_codec(Codec::Uuid);
         let mut decoder = Decoder::try_new(&avro_type).expect("Failed to create decoder");
-
-        let data1 = [1u8, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
+        let uuid_bytes = [1u8, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
+        let data1 = encode_avro_bytes(&uuid_bytes);
         let mut cursor1 = AvroCursor::new(&data1);
         decoder
             .decode(&mut cursor1)
             .expect("Failed to decode data1");
         assert_eq!(
             cursor1.position(),
-            16,
-            "Cursor should advance by fixed size"
+            17,
+            "Cursor should advance by varint size + data size"
         );
     }
 
@@ -1035,4 +1057,67 @@ mod tests {
         assert_eq!(values.value(0), "X");
         assert_eq!(values.value(1), "Y");
     }
+
+    #[test]
+    fn test_duration_decoding_with_nulls() {
+        let duration_codec = Codec::Interval;
+        let avro_type = AvroDataType::new(
+            duration_codec,
+            Default::default(),
+            Some(Nullability::NullFirst),
+        );
+        let mut decoder = Decoder::try_new(&avro_type).unwrap();
+        let mut data = Vec::new();
+        // First value: 1 month, 2 days, 3 millis
+        data.extend_from_slice(&encode_avro_long(1)); // not null
+        let mut duration1 = Vec::new();
+        duration1.extend_from_slice(&1u32.to_le_bytes());
+        duration1.extend_from_slice(&2u32.to_le_bytes());
+        duration1.extend_from_slice(&3u32.to_le_bytes());
+        data.extend_from_slice(&duration1);
+        // Second value: null
+        data.extend_from_slice(&encode_avro_long(0)); // null
+        data.extend_from_slice(&encode_avro_long(1)); // not null
+        let mut duration2 = Vec::new();
+        duration2.extend_from_slice(&4u32.to_le_bytes());
+        duration2.extend_from_slice(&5u32.to_le_bytes());
+        duration2.extend_from_slice(&6u32.to_le_bytes());
+        data.extend_from_slice(&duration2);
+        let mut cursor = AvroCursor::new(&data);
+        decoder.decode(&mut cursor).unwrap();
+        decoder.decode(&mut cursor).unwrap();
+        decoder.decode(&mut cursor).unwrap();
+        let array = decoder.flush(None).unwrap();
+        let interval_array = array
+            .as_any()
+            .downcast_ref::<IntervalMonthDayNanoArray>()
+            .unwrap();
+        assert_eq!(interval_array.len(), 3);
+        assert!(interval_array.is_valid(0));
+        assert!(interval_array.is_null(1));
+        assert!(interval_array.is_valid(2));
+        let expected = IntervalMonthDayNanoArray::from(vec![
+            Some(IntervalMonthDayNano {
+                months: 1,
+                days: 2,
+                nanoseconds: 3_000_000,
+            }),
+            None,
+            Some(IntervalMonthDayNano {
+                months: 4,
+                days: 5,
+                nanoseconds: 6_000_000,
+            }),
+        ]);
+        assert_eq!(interval_array, &expected);
+    }
+
+    #[test]
+    fn test_duration_decoding_empty() {
+        let duration_codec = Codec::Interval;
+        let avro_type = AvroDataType::new(duration_codec, Default::default(), None);
+        let mut decoder = Decoder::try_new(&avro_type).unwrap();
+        let array = decoder.flush(None).unwrap();
+        assert_eq!(array.len(), 0);
+    }
 }