AVRO-3892: [Rust] Support to resolve fixed from bytes and deserialize bytes in deserialize_any

lang/rust/avro/examples/benchmark.rs

@@ -60,7 +60,7 @@ fn benchmark(
 
         let start = Instant::now();
         let mut writer = Writer::new(schema, BufWriter::new(Vec::new()));
-        writer.extend(records.into_iter())?;
+        writer.extend(records)?;
 
         let duration = Instant::now().duration_since(start);
         durations.push(duration);

lang/rust/avro/src/de.rs

@@ -266,6 +266,8 @@ impl<'a, 'de> de::Deserializer<'de> for &'a Deserializer<'de> {
                 Value::String(ref s) => visitor.visit_borrowed_str(s),
                 Value::Uuid(uuid) => visitor.visit_str(&uuid.to_string()),
                 Value::Map(ref items) => visitor.visit_map(MapDeserializer::new(items)),
+                Value::Bytes(ref bytes) | Value::Fixed(_, ref bytes) => visitor.visit_bytes(bytes),
+                Value::Decimal(ref d) => visitor.visit_bytes(&d.to_vec()?),
                 _ => Err(de::Error::custom(format!(
                     "unsupported union: {:?}",
                     self.input
@@ -276,6 +278,8 @@ impl<'a, 'de> de::Deserializer<'de> for &'a Deserializer<'de> {
             Value::String(ref s) => visitor.visit_borrowed_str(s),
             Value::Uuid(uuid) => visitor.visit_str(&uuid.to_string()),
             Value::Map(ref items) => visitor.visit_map(MapDeserializer::new(items)),
+            Value::Bytes(ref bytes) | Value::Fixed(_, ref bytes) => visitor.visit_bytes(bytes),
+            Value::Decimal(ref d) => visitor.visit_bytes(&d.to_vec()?),
             value => Err(de::Error::custom(format!(
                 "incorrect value of type: {:?}",
                 crate::schema::SchemaKind::from(value)
@@ -350,6 +354,7 @@ impl<'a, 'de> de::Deserializer<'de> for &'a Deserializer<'de> {
             Value::String(ref s) => visitor.visit_bytes(s.as_bytes()),
             Value::Bytes(ref bytes) | Value::Fixed(_, ref bytes) => visitor.visit_bytes(bytes),
             Value::Uuid(ref u) => visitor.visit_bytes(u.as_bytes()),
+            Value::Decimal(ref d) => visitor.visit_bytes(&d.to_vec()?),
             _ => Err(de::Error::custom(format!(
                 "Expected a String|Bytes|Fixed|Uuid, but got {:?}",
                 self.input

lang/rust/avro/src/decimal.rs

@@ -44,7 +44,7 @@ impl Decimal {
         self.len
     }
 
-    fn to_vec(&self) -> AvroResult<Vec<u8>> {
+    pub(crate) fn to_vec(&self) -> AvroResult<Vec<u8>> {
         self.to_sign_extended_bytes_with_len(self.len)
     }
 

lang/rust/avro/src/types.rs

@@ -917,6 +917,13 @@ impl Value {
                 }
             }
             Value::String(s) => Ok(Value::Fixed(s.len(), s.into_bytes())),
+            Value::Bytes(s) => {
+                if s.len() == size {
+                    Ok(Value::Fixed(size, s))
+                } else {
+                    Err(Error::CompareFixedSizes { size, n: s.len() })
+                }
+            }
             other => Err(Error::GetStringForFixed(other.into())),
         }
     }
@@ -2958,4 +2965,43 @@ Field with name '"b"' is not a member of the map items"#,
 
         Ok(())
     }
+
+    #[test]
+    fn resolve_fixed_from_bytes() -> TestResult {
+        let value = Value::Bytes(vec![97, 98, 99]);
+        assert_eq!(
+            value.resolve(&Schema::Fixed(FixedSchema {
+                name: "test".into(),
+                aliases: None,
+                doc: None,
+                size: 3,
+                attributes: Default::default()
+            }))?,
+            Value::Fixed(3, vec![97, 98, 99])
+        );
+
+        let value = Value::Bytes(vec![97, 99]);
+        assert!(value
+            .resolve(&Schema::Fixed(FixedSchema {
+                name: "test".into(),
+                aliases: None,
+                doc: None,
+                size: 3,
+                attributes: Default::default()
+            }))
+            .is_err(),);
+
+        let value = Value::Bytes(vec![97, 98, 99, 100]);
+        assert!(value
+            .resolve(&Schema::Fixed(FixedSchema {
+                name: "test".into(),
+                aliases: None,
+                doc: None,
+                size: 3,
+                attributes: Default::default()
+            }))
+            .is_err(),);
+
+        Ok(())
+    }
 }

lang/rust/avro/src/writer.rs

@@ -895,7 +895,7 @@ mod tests {
         let record_copy = record.clone();
         let records = vec![record, record_copy];
 
-        let n1 = writer.extend(records.into_iter())?;
+        let n1 = writer.extend(records)?;
         let n2 = writer.flush()?;
         let result = writer.into_inner()?;
 
@@ -970,7 +970,7 @@ mod tests {
         let record_copy = record.clone();
         let records = vec![record, record_copy];
 
-        let n1 = writer.extend_ser(records.into_iter())?;
+        let n1 = writer.extend_ser(records)?;
         let n2 = writer.flush()?;
         let result = writer.into_inner()?;