From da98719ad84bad1ef71c17779ba6b3c482d84b2c Mon Sep 17 00:00:00 2001
From: Martin Grigorov <martin-g@users.noreply.github.com>
Date: Tue, 7 Nov 2023 14:49:15 +0200
Subject: [PATCH] AVRO-3899: [Rust] Invalid logical types should be ignored and
 treatedas the underlying type (#2580)

* AVRO-3899: [Rust] Invalid logical types should be ignored and treated as the underlying type

Log a warning for invalid logical (decimal) schema and use the
underlying (base) schema

Signed-off-by: Martin Tzvetanov Grigorov <mgrigorov@apache.org>

* AVRO-3899: [Rust] Add a new unit test for parsing decimal schema

Assert that a log message is logged for invalid schema.
Assert that the same log message is not logged for a valid decimal
schema.

Signed-off-by: Martin Tzvetanov Grigorov <mgrigorov@apache.org>

---------

Signed-off-by: Martin Tzvetanov Grigorov <mgrigorov@apache.org>
---
 lang/rust/avro/src/schema.rs   | 64 ++++++++++++++++++++++++++++++----
 lang/rust/avro/tests/schema.rs | 12 +++----
 2 files changed, 64 insertions(+), 12 deletions(-)

diff --git a/lang/rust/avro/src/schema.rs b/lang/rust/avro/src/schema.rs
index 262aa2affc1..bc371b5e00b 100644
--- a/lang/rust/avro/src/schema.rs
+++ b/lang/rust/avro/src/schema.rs
@@ -1324,12 +1324,17 @@ impl Parser {
                         parse_as_native_complex(complex, self, enclosing_namespace)?,
                         &[SchemaKind::Fixed, SchemaKind::Bytes],
                         |inner| -> AvroResult<Schema> {
-                            let (precision, scale) = Self::parse_precision_and_scale(complex)?;
-                            Ok(Schema::Decimal(DecimalSchema {
-                                precision,
-                                scale,
-                                inner: Box::new(inner),
-                            }))
+                            match Self::parse_precision_and_scale(complex) {
+                                Ok((precision, scale)) => Ok(Schema::Decimal(DecimalSchema {
+                                    precision,
+                                    scale,
+                                    inner: Box::new(inner),
+                                })),
+                                Err(err) => {
+                                    warn!("Ignoring invalid decimal logical type: {}", err);
+                                    Ok(inner)
+                                }
+                            }
                         },
                     );
                 }
@@ -6311,4 +6316,51 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_avro_3899_parse_decimal_type() -> TestResult {
+        use apache_avro_test_helper::logger::{assert_logged, assert_not_logged};
+
+        let schema = Schema::parse_str(
+            r#"{
+             "name": "InvalidDecimal",
+             "type": "fixed",
+             "size": 16,
+             "logicalType": "decimal",
+             "precision": 2,
+             "scale": 3
+         }"#,
+        )?;
+        match schema {
+            Schema::Fixed(fixed_schema) => {
+                let attrs = fixed_schema.attributes;
+                let precision = attrs
+                    .get("precision")
+                    .expect("The 'precision' attribute is missing");
+                let scale = attrs
+                    .get("scale")
+                    .expect("The 'scale' attribute is missing");
+                assert_logged(&format!("Ignoring invalid decimal logical type: The decimal precision ({}) must be bigger or equal to the scale ({})", precision, scale));
+            }
+            _ => unreachable!("Expected Schema::Fixed, got {:?}", schema),
+        }
+
+        let schema = Schema::parse_str(
+            r#"{
+            "name": "ValidDecimal",
+             "type": "bytes",
+             "logicalType": "decimal",
+             "precision": 3,
+             "scale": 2
+         }"#,
+        )?;
+        match schema {
+            Schema::Decimal(_) => {
+                assert_not_logged("Ignoring invalid decimal logical type: The decimal precision (2) must be bigger or equal to the scale (3)");
+            }
+            _ => unreachable!("Expected Schema::Decimal, got {:?}", schema),
+        }
+
+        Ok(())
+    }
 }
diff --git a/lang/rust/avro/tests/schema.rs b/lang/rust/avro/tests/schema.rs
index 63b73056084..d548281da90 100644
--- a/lang/rust/avro/tests/schema.rs
+++ b/lang/rust/avro/tests/schema.rs
@@ -431,7 +431,7 @@ const DECIMAL_LOGICAL_TYPE: &[(&str, bool)] = &[
             "logicalType": "decimal",
             "precision": 0
         }"#,
-        false,
+        true,
     ),
     (
         r#"{
@@ -449,7 +449,7 @@ const DECIMAL_LOGICAL_TYPE: &[(&str, bool)] = &[
              "precision": 2,
              "scale": -2
          }"#,
-        false,
+        true,
     ),
     (
         r#"{
@@ -458,7 +458,7 @@ const DECIMAL_LOGICAL_TYPE: &[(&str, bool)] = &[
              "precision": -2,
              "scale": 2
          }"#,
-        false,
+        true,
     ),
     (
         r#"{
@@ -467,7 +467,7 @@ const DECIMAL_LOGICAL_TYPE: &[(&str, bool)] = &[
              "precision": 2,
              "scale": 3
          }"#,
-        false,
+        true,
     ),
     (
         r#"{
@@ -478,7 +478,7 @@ const DECIMAL_LOGICAL_TYPE: &[(&str, bool)] = &[
              "scale": 2,
              "size": 5
          }"#,
-        false,
+        true,
     ),
     (
         r#"{
@@ -489,7 +489,7 @@ const DECIMAL_LOGICAL_TYPE: &[(&str, bool)] = &[
              "scale": 3,
              "size": 2
          }"#,
-        false,
+        true,
     ),
     (
         r#"{