AVRO-3912: Fix Big decimal deser

lang/rust/avro/src/decimal.rs

@@ -113,13 +113,17 @@ impl<T: AsRef<[u8]>> From<T> for Decimal {
 }
 
 pub(crate) fn serialize_big_decimal(decimal: &BigDecimal) -> Vec<u8> {
+    // encode big decimal, without global size
     let mut buffer: Vec<u8> = Vec::new();
     let (big_int, exponent): (BigInt, i64) = decimal.as_bigint_and_exponent();
     let big_endian_value: Vec<u8> = big_int.to_signed_bytes_be();
     encode_bytes(&big_endian_value, &mut buffer);
     encode_long(exponent, &mut buffer);
 
-    buffer
+    // encode global size and content
+    let mut final_buffer: Vec<u8> = Vec::new();
+    encode_bytes(&buffer, &mut final_buffer);
+    final_buffer
 }
 
 pub(crate) fn deserialize_big_decimal(bytes: &Vec<u8>) -> Result<BigDecimal, Error> {
@@ -146,12 +150,19 @@ pub(crate) fn deserialize_big_decimal(bytes: &Vec<u8>) -> Result<BigDecimal, Err
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::{
+        types::{Record, Value},
+        Codec, Reader, Schema, Writer,
+    };
     use apache_avro_test_helper::TestResult;
     use bigdecimal::{One, Zero};
     use pretty_assertions::assert_eq;
     use std::{
         convert::TryFrom,
+        fs::File,
+        io::BufReader,
         ops::{Div, Mul},
+        str::FromStr,
     };
 
     #[test]
@@ -178,40 +189,126 @@ mod tests {
 
     #[test]
     fn test_avro_3779_bigdecimal_serial() -> TestResult {
-        let value: bigdecimal::BigDecimal =
+        let value: BigDecimal =
             bigdecimal::BigDecimal::from(-1421).div(bigdecimal::BigDecimal::from(2));
-        let mut current: bigdecimal::BigDecimal = bigdecimal::BigDecimal::one();
+        let mut current: BigDecimal = BigDecimal::one();
 
         for iter in 1..180 {
-            let result: Vec<u8> = serialize_big_decimal(&current);
+            let buffer: Vec<u8> = serialize_big_decimal(&current);
+
+            let mut as_slice = buffer.as_slice();
+            decode_long(&mut as_slice)?;
 
-            let deserialize_big_decimal: Result<bigdecimal::BigDecimal, Error> =
+            let mut result: Vec<u8> = Vec::new();
+            result.extend_from_slice(as_slice);
+
+            let deserialize_big_decimal: Result<BigDecimal, Error> =
                 deserialize_big_decimal(&result);
             assert!(
                 deserialize_big_decimal.is_ok(),
                 "can't deserialize for iter {iter}"
             );
-            assert_eq!(
-                current,
-                deserialize_big_decimal.unwrap(),
-                "not equals for ${iter}"
-            );
+            assert_eq!(current, deserialize_big_decimal?, "not equals for {iter}");
             current = current.mul(&value);
         }
 
-        let result: Vec<u8> = serialize_big_decimal(&BigDecimal::zero());
-        let deserialize_big_decimal: Result<bigdecimal::BigDecimal, Error> =
-            deserialize_big_decimal(&result);
+        let buffer: Vec<u8> = serialize_big_decimal(&BigDecimal::zero());
+        let mut as_slice = buffer.as_slice();
+        decode_long(&mut as_slice)?;
+
+        let mut result: Vec<u8> = Vec::new();
+        result.extend_from_slice(as_slice);
+
+        let deserialize_big_decimal: Result<BigDecimal, Error> = deserialize_big_decimal(&result);
         assert!(
             deserialize_big_decimal.is_ok(),
             "can't deserialize for zero"
         );
         assert_eq!(
             BigDecimal::zero(),
-            deserialize_big_decimal.unwrap(),
+            deserialize_big_decimal?,
             "not equals for zero"
         );
 
         Ok(())
     }
+
+    #[test]
+    fn test_avro_3779_record_with_bg() -> TestResult {
+        let schema_str = r#"
+        {
+          "type": "record",
+          "name": "test",
+          "fields": [
+            {
+              "name": "field_name",
+              "type": "bytes",
+              "logicalType": "big-decimal"
+            }
+          ]
+        }
+        "#;
+        let schema = Schema::parse_str(schema_str)?;
+
+        // build record with big decimal value
+        let mut record = Record::new(&schema)
+            .expect(format!("Cannot create a Record from {schema_str}").as_str());
+        let val = BigDecimal::new(BigInt::from(12), 2);
+        record.put("field_name", val.clone());
+
+        // write a record
+        let codec = Codec::Null;
+        let mut writer = Writer::builder()
+            .schema(&schema)
+            .codec(codec)
+            .writer(Vec::new())
+            .build();
+
+        writer.append(record.clone())?;
+        writer.flush()?;
+
+        // read record
+        let wrote_data = writer.into_inner()?;
+        let mut reader = Reader::new(&wrote_data[..])?;
+
+        let value = reader.next().unwrap()?;
+
+        // extract field value
+        let big_decimal_value: &Value = match value {
+            Value::Record(ref fields) => Ok(&fields[0].1),
+            other => Err(format!("Expected a Value::Record, got: {other:?}")),
+        }?;
+
+        let x1res: &BigDecimal = match big_decimal_value {
+            Value::BigDecimal(ref s) => Ok(s),
+            other => Err(format!("Expected Value::BigDecimal, got: {other:?}")),
+        }?;
+        assert_eq!(&val, x1res);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_avro_3779_from_java_file() -> TestResult {
+        // Open file generated with Java code to ensure compatibility
+        // with Java big decimal logical type.
+        let file: File = File::open("./tests/bigdec.avro")?;
+        let mut reader = Reader::new(BufReader::new(&file))?;
+        let next_element = reader.next();
+        assert!(next_element.is_some());
+        let value = next_element.unwrap()?;
+        let bg = match value {
+            Value::Record(ref fields) => Ok(&fields[0].1),
+            other => Err(format!("Expected a Value::Record, got: {other:?}")),
+        }?;
+        let value_big_decimal = match bg {
+            Value::BigDecimal(val) => Ok(val),
+            other => Err(format!("Expected a Value::BigDecimal, got: {other:?}")),
+        }?;
+
+        let ref_value = BigDecimal::from_str("2.24")?;
+        assert_eq!(&ref_value, value_big_decimal);
+
+        Ok(())
+    }
 }

lang/rust/avro/src/encode.rs

@@ -130,8 +130,8 @@ pub(crate) fn encode_internal<S: Borrow<Schema>>(
             buffer,
         ),
         Value::BigDecimal(bg) => {
-            let mut buf: Vec<u8> = serialize_big_decimal(bg);
-            buffer.append(&mut buf);
+            let buf: Vec<u8> = serialize_big_decimal(bg);
+            buffer.extend_from_slice(buf.as_slice());
         }
         Value::Bytes(bytes) => match *schema {
             Schema::Bytes => encode_bytes(bytes, buffer),