feat!: Support enums and tuples in SchemaAwareSerializer and implement SchemaAwareDeserializer (#512)

* doc: Document the mapping between the Serde and Avro data models

* feat: `SchemaAwareDeserializer` (no tests)

* feat: Use `SchemaAwareDeserializer` in the readers and add tests

* feat: Rework `SchemaAwareSerializer` to be more strict and follow the documented model (no tests)

* feat: Update tests and provide support for old ways of `BigDecimal` and `[T; N]`

* fix: Add missing license header and set PR number for tests

* fix: Don't call `T::field_default()` if the schema is overwritten using `#[avro(with)]`

* fix: Enable `rustdoc_internal` feature when `cfg(docsrs)` so we can use `fake_variadic`

* fix: Review feedback

---------
Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com>

* fix: Apply suggestions from code review

Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com>

* fix: Second round of code review

Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com>

* fix: Use longs when serializing block item count and byte size

* fix: third round of review feedback

Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com>

* fix: Don't allow serializing a unit variant inside the UnionSerializer as that would require a nested union

* chore: Enable and fix Clippy lints

* fix: fourth round of review feedback

* fix: Typo

Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com>

---------

Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com>

Author: Kriskras99

Cargo.toml

@@ -58,14 +58,13 @@ clippy.too_long_first_doc_paragraph = "warn"
 clippy.doc_markdown = "warn"
 # TODO: Needs more work
 #clippy.missing_errors_doc = "warn"
-#clippy.missing_panics_doc = "warn"
+clippy.missing_panics_doc = "warn"
 rust.unexpected_cfgs = { level = "warn", check-cfg = ['cfg(nightly)'] }
 clippy.cargo = { level = "warn", priority = -1 }
 clippy.multiple_crate_versions = "allow"
-# clippy.needless_raw_string_hashes = "warn"
-# clippy.missing_panics_doc = "warn"
-# clippy.semicolon_if_nothing_returned = "warn"
-# clippy.manual_assert = "warn"
-# clippy.enum_glob_use = "warn"
-# clippy.needless_pass_by_value = "warn"
-# clippy.single_match_else = "warn"
+clippy.needless_raw_string_hashes = "warn"
+clippy.semicolon_if_nothing_returned = "warn"
+clippy.manual_assert = "warn"
+clippy.enum_glob_use = "warn"
+clippy.needless_pass_by_value = "warn"
+clippy.single_match_else = "warn"

avro/benches/serde.rs

@@ -360,7 +360,7 @@ fn bench_small_schema_write_10_000_record_ser(c: &mut Criterion) {
         10_000,
         "small schema, write 10k records (serde way)",
     )
-    .unwrap()
+    .unwrap();
 }
 
 fn bench_small_schema_read_1_record(c: &mut Criterion) {

avro/benches/single.rs

@@ -180,7 +180,7 @@ fn bench_small_schema_write_record(c: &mut Criterion) {
                 .build()
                 .unwrap()
                 .write_value_to_vec(record.clone())
-        })
+        });
     });
 }
 
@@ -192,23 +192,23 @@ fn bench_big_schema_write_record(c: &mut Criterion) {
                 .build()
                 .unwrap()
                 .write_value_to_vec(record.clone())
-        })
+        });
     });
 }
 
 fn bench_small_schema_write_record_reuse_datum_writer(c: &mut Criterion) {
     let (schema, record) = make_small_record().unwrap();
     let writer = GenericDatumWriter::builder(&schema).build().unwrap();
     c.bench_function("small record (reused writer)", |b| {
-        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record))
+        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record));
     });
 }
 
 fn bench_big_schema_write_record_reuse_datum_writer(c: &mut Criterion) {
     let (schema, record) = make_big_record().unwrap();
     let writer = GenericDatumWriter::builder(&schema).build().unwrap();
     c.bench_function("big record (reused writer)", |b| {
-        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record))
+        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record));
     });
 }
 
@@ -219,7 +219,7 @@ fn bench_small_schema_write_record_no_validation(c: &mut Criterion) {
         .build()
         .unwrap();
     c.bench_function("small record (no validation)", |b| {
-        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record))
+        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record));
     });
 }
 
@@ -230,7 +230,7 @@ fn bench_big_schema_write_record_no_validation(c: &mut Criterion) {
         .build()
         .unwrap();
     c.bench_function("big record (no validation)", |b| {
-        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record))
+        b.iter(|| writer.write_value_ref(&mut Vec::new(), &record));
     });
 }
 

avro/examples/test_interop_single_object_encoding.rs

@@ -58,7 +58,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     let single_object = std::fs::read(format!("{resource_folder}/test_message.bin"))
         .expect("File with single object not found or error occurred while reading it.");
     test_write(&single_object);
-    test_read(single_object);
+    test_read(&single_object);
 
     Ok(())
 }
@@ -69,15 +69,16 @@ fn test_write(expected: &[u8]) {
         .expect("Resolving failed")
         .write_value(InteropMessage, &mut encoded)
         .expect("Encoding failed");
-    assert_eq!(expected, &encoded)
+    assert_eq!(expected, &encoded);
 }
 
-fn test_read(encoded: Vec<u8>) {
-    let mut encoded = &encoded[..];
-    let read_message = apache_avro::GenericSingleObjectReader::new(InteropMessage::get_schema())
+fn test_read(mut encoded: &[u8]) {
+    let read_message = apache_avro::GenericSingleObjectReader::builder()
+        .schema(InteropMessage::get_schema())
+        .build()
         .expect("Resolving failed")
         .read_value(&mut encoded)
         .expect("Decoding failed");
     let expected_value: Value = InteropMessage.into();
-    assert_eq!(expected_value, read_message)
+    assert_eq!(expected_value, read_message);
 }

avro/src/bigdecimal.rs

@@ -69,18 +69,23 @@ pub(crate) fn deserialize_big_decimal(mut bytes: &[u8]) -> AvroResult<BigDecimal
 
 #[cfg(test)]
 mod tests {
-    use super::*;
-    use crate::{Codec, Reader, Schema, Writer, error::Error, from_value, types::Record};
-    use apache_avro_test_helper::TestResult;
-    use bigdecimal::{One, Zero};
-    use pretty_assertions::assert_eq;
     use std::{
         fs::File,
         io::BufReader,
         ops::{Div, Mul},
         str::FromStr,
     };
 
+    use apache_avro_test_helper::TestResult;
+    use bigdecimal::{One, Zero};
+    use pretty_assertions::assert_eq;
+
+    use super::*;
+    use crate::{
+        Codec, Reader, Schema, Writer, error::Error, reader::datum::GenericDatumReader,
+        types::Record, writer::datum::GenericDatumWriter,
+    };
+
     #[test]
     fn test_avro_3779_bigdecimal_serial() -> TestResult {
         let value: BigDecimal =
@@ -190,6 +195,43 @@ mod tests {
             big_decimal: BigDecimal,
         }
 
+        let schema_str = r#"
+        {
+          "type": "record",
+          "name": "Test",
+          "fields": [
+            {
+              "name": "big_decimal",
+              "type": "string"
+            }
+          ]
+        }
+        "#;
+        let schema = Schema::parse_str(schema_str)?;
+
+        let test = Test::default();
+
+        let serialized = GenericDatumWriter::builder(&schema)
+            .build()?
+            .write_ser_to_vec(&test)?;
+        let value: Test = GenericDatumReader::builder(&schema)
+            .build()?
+            .read_deser(&mut &serialized[..])?;
+
+        assert_eq!(value, test);
+
+        Ok(())
+    }
+
+    #[test]
+    fn avro_rs_338_deserialize_serde_way_with_bigdecimal() -> TestResult {
+        #[derive(Clone, PartialEq, Eq, Debug, Default, serde::Deserialize, serde::Serialize)]
+        #[serde(rename = "test")]
+        struct Test {
+            #[serde(with = "crate::serde::bigdecimal")]
+            big_decimal: BigDecimal,
+        }
+
         let schema_str = r#"
         {
           "type": "record",
@@ -216,11 +258,11 @@ mod tests {
         let wrote_data = writer.into_inner()?;
 
         // read record
-        let mut reader = Reader::new(&wrote_data[..])?;
+        let mut reader = Reader::new(&wrote_data[..])?.into_deser_iter();
 
         let value = reader.next().unwrap()?;
 
-        assert_eq!(test, from_value::<Test>(&value)?);
+        assert_eq!(test, value);
 
         Ok(())
     }

avro/src/codec.rs

@@ -108,10 +108,10 @@ impl Codec {
             #[cfg(feature = "zstandard")]
             Codec::Zstandard(settings) => {
                 use std::io::Write;
-                let mut encoder =
-                    zstd::Encoder::new(Vec::new(), settings.compression_level as i32).unwrap();
+                let mut encoder = zstd::Encoder::new(Vec::new(), settings.compression_level as i32)
+                    .map_err(Details::ZstdCompress)?;
                 encoder.write_all(stream).map_err(Details::ZstdCompress)?;
-                *stream = encoder.finish().unwrap();
+                *stream = encoder.finish().map_err(Details::ZstdCompress)?;
             }
             #[cfg(feature = "bzip")]
             Codec::Bzip2(settings) => {
@@ -120,7 +120,9 @@ impl Codec {
 
                 let mut encoder = BzEncoder::new(&stream[..], settings.compression());
                 let mut buffer = Vec::new();
-                encoder.read_to_end(&mut buffer).unwrap();
+                encoder
+                    .read_to_end(&mut buffer)
+                    .unwrap_or_else(|_| unreachable!("No I/O errors possible with Vec<u8>"));
                 *stream = buffer;
             }
             #[cfg(feature = "xz")]
@@ -130,7 +132,9 @@ impl Codec {
 
                 let mut encoder = XzEncoder::new(&stream[..], settings.compression_level as u32);
                 let mut buffer = Vec::new();
-                encoder.read_to_end(&mut buffer).unwrap();
+                encoder
+                    .read_to_end(&mut buffer)
+                    .unwrap_or_else(|_| unreachable!("No I/O errors possible with Vec<u8>"));
                 *stream = buffer;
             }
         };
@@ -144,7 +148,7 @@ impl Codec {
             Codec::Null => return Ok(()),
             Codec::Deflate(_settings) => miniz_oxide::inflate::decompress_to_vec(stream).map_err(|e| {
                 let err = {
-                    use miniz_oxide::inflate::TINFLStatus::*;
+                    use miniz_oxide::inflate::TINFLStatus::{FailedCannotMakeProgress, BadParam, Adler32Mismatch, Failed, Done, NeedsMoreInput, HasMoreOutput};
                     use std::io::{Error,ErrorKind};
                     match e.status {
                         FailedCannotMakeProgress => Error::from(ErrorKind::UnexpectedEof),
@@ -189,7 +193,7 @@ impl Codec {
                 let mut decoded = Vec::new();
                 let buffer_size = zstd_safe::DCtx::in_size();
                 let buffer = BufReader::with_capacity(buffer_size, &stream[..]);
-                let mut decoder = zstd::Decoder::new(buffer).unwrap();
+                let mut decoder = zstd::Decoder::new(buffer).map_err(Details::ZstdDecompress)?;
                 std::io::copy(&mut decoder, &mut decoded).map_err(Details::ZstdDecompress)?;
                 decoded
             }
@@ -200,7 +204,7 @@ impl Codec {
 
                 let mut decoder = BzDecoder::new(&stream[..]);
                 let mut decoded = Vec::new();
-                decoder.read_to_end(&mut decoded).unwrap();
+                decoder.read_to_end(&mut decoded).unwrap_or_else(|_| unreachable!("No I/O errors possible with Vec<u8>"));
                 decoded
             }
             #[cfg(feature = "xz")]
@@ -210,7 +214,7 @@ impl Codec {
 
                 let mut decoder = XzDecoder::new(&stream[..]);
                 let mut decoded: Vec<u8> = Vec::new();
-                decoder.read_to_end(&mut decoded).unwrap();
+                decoder.read_to_end(&mut decoded).unwrap_or_else(|_| unreachable!("No I/O errors possible with Vec<u8>"));
                 decoded
             }
         };

avro/src/documentation/avro_data_model_to_serde.rs

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! # Mapping the Avro data model to the Serde data model
+//!
+//! When manually mapping an Avro schema to Rust types it is important to understand
+//! how the different data models are mapped. When mapping from Rust types to an Avro schema,
+//! see [the documentation for the reverse](super::serde_data_model_to_avro).
+//!
+//! Only the mapping as defined here is supported. Any other behavior might change in a minor version.
+//!
+//! ## Primitive Types
+//! - `null`: `()`
+//! - `boolean`: [`bool`]
+//! - `int`: [`i32`] (or [`i16`], [`i8`], [`u16`], [`u8`])
+//! - `long`: [`i64`] (or [`u32`])
+//! - `float`: [`f32`]
+//! - `double`: [`f64`]
+//! - `bytes`: [`Vec<u8>`](std::vec::Vec) (or any type that uses [`Serializer::serialize_bytes`](serde::Serializer), [`Deserializer::deserialize_bytes`](serde::Deserializer), [`Deserializer::deserialize_byte_buf`](serde::Deserializer))
+//!     - It is required to use [`apache_avro::serde::bytes`] as otherwise Serde will (de)serialize a `Vec` as an array of integers instead.
+//! - `string`: [`String`] (or any type that uses [`Serializer::serialize_str`](serde::Serializer), [`Deserializer::deserialize_str`](serde::Deserializer), [`Deserializer::deserialize_string`](serde::Deserializer))
+//!
+//! ## Complex Types
+//! - `records`: A struct with the same name and fields or a tuple with the same fields.
+//!     - Extra fields can be added to the struct if they are marked with `#[serde(skip)]`
+//! - `enums`: A enum with the same name and unit variants for every symbol
+//!     - The index of the symbol must match the index of the variant
+//! - `arrays`: [`Vec`] (or any type that uses [`Serializer::serialize_seq`](serde::Serializer), [`Deserializer::deserialize_seq`](serde::Deserializer))
+//!     - `[T; N]` is (de)serialized as a tuple by Serde, to (de)serialize them as an `array` use [`apache_avro::serde::array`]
+//! - `maps`: [`HashMap<String, _>`](std::collections::HashMap) (or any type that uses [`Serializer::serialize_map`](serde::Serializer), [`Deserializer::deserialize_map`](serde::Deserializer))
+//! - `unions`: An enum with a variant for each union variant
+//!     - The index of the union variant must match the enum variant
+//!     - A `null` can be a unit variant or a newtype variant with a unit type
+//!     - All other variants must be newtype variants, struct variants, or tuple variants.
+//! - `fixed`: [`Vec<u8>`](std::vec::Vec) (or any type that uses [`Serializer::serialize_bytes`](serde::Serializer), [`Deserializer::deserialize_bytes`](serde::Deserializer), [`Deserializer::deserialize_byte_buf`](serde::Deserializer))
+//!     - It is required to use [`apache_avro::serde::fixed`] as otherwise Serde will (de)serialize a `Vec` as an array of integers instead.
+//!
+//! [`apache_avro::serde::array`]: crate::serde::array
+//! [`apache_avro::serde::bytes`]: crate::serde::bytes
+//! [`apache_avro::serde::fixed`]: crate::serde::fixed

avro/src/documentation/mod.rs

@@ -20,5 +20,7 @@
 //! This module does not contain any code, and is only available during `rustdoc` builds.
 //!
 
+pub mod avro_data_model_to_serde;
 pub mod dynamic;
 pub mod primer;
+pub mod serde_data_model_to_avro;