feat: Implement support for #[serde(flatten)] (#359)

* chore: Move all Serde related modules to the `serde` module

* feat: Implement support for `#[serde(flatten)]`

This is done by adding a `#[avro(flatten]` attribute so that the
schema (for that field) is also flattened, and by adding support
in `SchemaAwareWriteSerializer` for serializing a struct via Map
instead of Struct.

`flatten` does not work with `to_value`, as `to_value` does not have
access to the schema.

* fix: Handle duplicate fields in when flatten is used

---------

Co-authored-by: default <admin@kriskras99.nl>

Author: Kriskras99

avro/src/error.rs

@@ -579,6 +579,12 @@ pub enum Details {
 
     #[error("Cannot convert a slice to Uuid: {0}")]
     UuidFromSlice(#[source] uuid::Error),
+
+    #[error("Expected String for Map key when serializing a flattened struct")]
+    MapFieldExpectedString,
+
+    #[error("No key for value when serializing a map")]
+    MapNoKey,
 }
 
 #[derive(thiserror::Error, PartialEq)]

avro/src/lib.rs

@@ -945,14 +945,12 @@
 mod bigdecimal;
 mod bytes;
 mod codec;
-mod de;
 mod decimal;
 mod decode;
 mod duration;
 mod encode;
 mod reader;
-mod ser;
-mod ser_schema;
+mod serde;
 mod writer;
 
 pub mod error;
@@ -979,7 +977,6 @@ pub use codec::xz::XzSettings;
 #[cfg(feature = "zstandard")]
 pub use codec::zstandard::ZstandardSettings;
 pub use codec::{Codec, DeflateSettings};
-pub use de::from_value;
 pub use decimal::Decimal;
 pub use duration::{Days, Duration, Millis, Months};
 pub use error::Error;
@@ -988,7 +985,7 @@ pub use reader::{
     from_avro_datum_reader_schemata, from_avro_datum_schemata, read_marker,
 };
 pub use schema::{AvroSchema, Schema};
-pub use ser::to_value;
+pub use serde::{de::from_value, ser::to_value};
 pub use uuid::Uuid;
 pub use writer::{
     GenericSingleObjectWriter, SpecificSingleObjectWriter, Writer, WriterBuilder, to_avro_datum,

avro/src/de.rs avro/src/serde/de.rsavro/src/de.rs renamed to avro/src/serde/de.rs

@@ -0,0 +1,4 @@
+pub mod de;
+pub mod ser;
+pub mod ser_schema;
+mod util;

avro/src/serde/mod.rs

@@ -479,7 +479,12 @@ impl ser::SerializeStructVariant for StructVariantSerializer<'_> {
 /// Interpret a serializeable instance as a `Value`.
 ///
 /// This conversion can fail if the value is not valid as per the Avro specification.
-/// e.g: HashMap with non-string keys
+/// e.g: `HashMap` with non-string keys.
+///
+/// This function does not work if `S` has any fields (recursively) that have the `#[serde(flatten)]`
+/// attribute. Please use [`Writer::append_ser`] if that's the case.
+///
+/// [`Writer::append_ser`]: crate::Writer::append_ser
 pub fn to_value<S: Serialize>(value: S) -> Result<Value, Error> {
     let mut serializer = Serializer::default();
     value.serialize(&mut serializer)

avro/src/ser.rs avro/src/serde/ser.rsavro/src/ser.rs renamed to avro/src/serde/ser.rs

@@ -23,9 +23,10 @@ use crate::{
     encode::{encode_int, encode_long},
     error::{Details, Error},
     schema::{Name, NamesRef, Namespace, RecordField, RecordSchema, Schema},
+    serde::util::StringSerializer,
 };
 use bigdecimal::BigDecimal;
-use serde::ser;
+use serde::{Serialize, ser};
 use std::{borrow::Cow, cmp::Ordering, collections::HashMap, io::Write, str::FromStr};
 
 const COLLECTION_SERIALIZER_ITEM_LIMIT: usize = 1024;
@@ -251,6 +252,8 @@ pub struct SchemaAwareWriteSerializeStruct<'a, 's, W: Write> {
     record_schema: &'s RecordSchema,
     /// Fields we received in the wrong order
     field_cache: HashMap<usize, Vec<u8>>,
+    /// The current field name when serializing from a map (for `flatten` support).
+    map_field_name: Option<String>,
     field_position: usize,
     bytes_written: usize,
 }
@@ -264,6 +267,7 @@ impl<'a, 's, W: Write> SchemaAwareWriteSerializeStruct<'a, 's, W> {
             ser,
             record_schema,
             field_cache: HashMap::new(),
+            map_field_name: None,
             field_position: 0,
             bytes_written: 0,
         }
@@ -352,6 +356,11 @@ impl<'a, 's, W: Write> SchemaAwareWriteSerializeStruct<'a, 's, W> {
             "There should be no more unwritten fields at this point: {:?}",
             self.field_cache
         );
+        debug_assert!(
+            self.map_field_name.is_none(),
+            "There should be no field name at this point: field {:?}",
+            self.map_field_name
+        );
         Ok(self.bytes_written)
     }
 }
@@ -371,17 +380,14 @@ impl<W: Write> ser::SerializeStruct for SchemaAwareWriteSerializeStruct<'_, '_,
             .and_then(|idx| self.record_schema.fields.get(*idx));
 
         match record_field {
-            Some(field) => {
-                // self.item_count += 1;
-                self.serialize_next_field(field, value).map_err(|e| {
-                    Details::SerializeRecordFieldWithSchema {
-                        field_name: key.to_string(),
-                        record_schema: Schema::Record(self.record_schema.clone()),
-                        error: Box::new(e),
-                    }
-                    .into()
-                })
-            }
+            Some(field) => self.serialize_next_field(field, value).map_err(|e| {
+                Details::SerializeRecordFieldWithSchema {
+                    field_name: key.to_string(),
+                    record_schema: Schema::Record(self.record_schema.clone()),
+                    error: Box::new(e),
+                }
+                .into()
+            }),
             None => Err(Details::FieldName(String::from(key)).into()),
         }
     }
@@ -420,6 +426,53 @@ impl<W: Write> ser::SerializeStruct for SchemaAwareWriteSerializeStruct<'_, '_,
     }
 }
 
+/// This implementation is used to support `#[serde(flatten)]` as that uses SerializeMap instead of SerializeStruct.
+impl<W: Write> ser::SerializeMap for SchemaAwareWriteSerializeStruct<'_, '_, W> {
+    type Ok = usize;
+    type Error = Error;
+
+    fn serialize_key<T>(&mut self, key: &T) -> Result<(), Self::Error>
+    where
+        T: ?Sized + Serialize,
+    {
+        let name = key.serialize(StringSerializer)?;
+        let old = self.map_field_name.replace(name);
+        debug_assert!(
+            old.is_none(),
+            "Expected a value instead of a key: old key: {old:?}, new key: {:?}",
+            self.map_field_name
+        );
+        Ok(())
+    }
+
+    fn serialize_value<T>(&mut self, value: &T) -> Result<(), Self::Error>
+    where
+        T: ?Sized + Serialize,
+    {
+        let key = self.map_field_name.take().ok_or(Details::MapNoKey)?;
+        let record_field = self
+            .record_schema
+            .lookup
+            .get(&key)
+            .and_then(|idx| self.record_schema.fields.get(*idx));
+        match record_field {
+            Some(field) => self.serialize_next_field(field, value).map_err(|e| {
+                Details::SerializeRecordFieldWithSchema {
+                    field_name: key.to_string(),
+                    record_schema: Schema::Record(self.record_schema.clone()),
+                    error: Box::new(e),
+                }
+                .into()
+            }),
+            None => Err(Details::FieldName(key).into()),
+        }
+    }
+
+    fn end(self) -> Result<Self::Ok, Self::Error> {
+        self.end()
+    }
+}
+
 impl<W: Write> ser::SerializeStructVariant for SchemaAwareWriteSerializeStruct<'_, '_, W> {
     type Ok = usize;
     type Error = Error;
@@ -436,6 +489,46 @@ impl<W: Write> ser::SerializeStructVariant for SchemaAwareWriteSerializeStruct<'
     }
 }
 
+/// Map serializer that switches between Struct or Map.
+///
+/// This exists because when `#[serde(flatten)]` is used, struct fields are serialized as a map.
+pub enum SchemaAwareWriteSerializeMapOrStruct<'a, 's, W: Write> {
+    Struct(SchemaAwareWriteSerializeStruct<'a, 's, W>),
+    Map(SchemaAwareWriteSerializeMap<'a, 's, W>),
+}
+
+impl<W: Write> ser::SerializeMap for SchemaAwareWriteSerializeMapOrStruct<'_, '_, W> {
+    type Ok = usize;
+    type Error = Error;
+
+    fn serialize_key<T>(&mut self, key: &T) -> Result<(), Self::Error>
+    where
+        T: ?Sized + Serialize,
+    {
+        match self {
+            Self::Struct(s) => s.serialize_key(key),
+            Self::Map(s) => s.serialize_key(key),
+        }
+    }
+
+    fn serialize_value<T>(&mut self, value: &T) -> Result<(), Self::Error>
+    where
+        T: ?Sized + Serialize,
+    {
+        match self {
+            Self::Struct(s) => s.serialize_value(value),
+            Self::Map(s) => s.serialize_value(value),
+        }
+    }
+
+    fn end(self) -> Result<Self::Ok, Self::Error> {
+        match self {
+            Self::Struct(s) => s.end(),
+            Self::Map(s) => s.end(),
+        }
+    }
+}
+
 /// The tuple struct serializer for [`SchemaAwareWriteSerializer`].
 /// [`SchemaAwareWriteSerializeTupleStruct`] can serialize to an Avro array, record, or big-decimal.
 /// When serializing to a record, fields must be provided in the correct order, since no names are provided.
@@ -1499,7 +1592,7 @@ impl<'s, W: Write> SchemaAwareWriteSerializer<'s, W> {
         &'a mut self,
         len: Option<usize>,
         schema: &'s Schema,
-    ) -> Result<SchemaAwareWriteSerializeMap<'a, 's, W>, Error> {
+    ) -> Result<SchemaAwareWriteSerializeMapOrStruct<'a, 's, W>, Error> {
         let create_error = |cause: String| {
             let len_str = len
                 .map(|l| format!("{l}"))
@@ -1513,15 +1606,17 @@ impl<'s, W: Write> SchemaAwareWriteSerializer<'s, W> {
         };
 
         match schema {
-            Schema::Map(map_schema) => Ok(SchemaAwareWriteSerializeMap::new(
-                self,
-                map_schema.types.as_ref(),
-                len,
+            Schema::Map(map_schema) => Ok(SchemaAwareWriteSerializeMapOrStruct::Map(
+                SchemaAwareWriteSerializeMap::new(self, map_schema.types.as_ref(), len),
             )),
+            Schema::Ref { name: ref_name } => {
+                let ref_schema = self.get_ref_schema(ref_name)?;
+                self.serialize_map_with_schema(len, ref_schema)
+            }
             Schema::Union(union_schema) => {
                 for (i, variant_schema) in union_schema.schemas.iter().enumerate() {
                     match variant_schema {
-                        Schema::Map(_) => {
+                        Schema::Map(_) | Schema::Record(_) | Schema::Ref { .. } => {
                             encode_int(i as i32, &mut *self.writer)?;
                             return self.serialize_map_with_schema(len, variant_schema);
                         }
@@ -1532,6 +1627,9 @@ impl<'s, W: Write> SchemaAwareWriteSerializer<'s, W> {
                     "Expected a Map schema in {union_schema:?}"
                 )))
             }
+            Schema::Record(record_schema) => Ok(SchemaAwareWriteSerializeMapOrStruct::Struct(
+                SchemaAwareWriteSerializeStruct::new(self, record_schema),
+            )),
             _ => Err(create_error(format!(
                 "Expected Map or Union schema. Got: {schema}"
             ))),
@@ -1630,7 +1728,7 @@ impl<'a, 's, W: Write> ser::Serializer for &'a mut SchemaAwareWriteSerializer<'s
     type SerializeTuple = SchemaAwareWriteSerializeSeq<'a, 's, W>;
     type SerializeTupleStruct = SchemaAwareWriteSerializeTupleStruct<'a, 's, W>;
     type SerializeTupleVariant = SchemaAwareWriteSerializeTupleStruct<'a, 's, W>;
-    type SerializeMap = SchemaAwareWriteSerializeMap<'a, 's, W>;
+    type SerializeMap = SchemaAwareWriteSerializeMapOrStruct<'a, 's, W>;
     type SerializeStruct = SchemaAwareWriteSerializeStruct<'a, 's, W>;
     type SerializeStructVariant = SchemaAwareWriteSerializeStruct<'a, 's, W>;