AVRO-3904: [RUST] return a Result when checking schema compatibility …

…so the … (#2587)

* AVRO-3904: [Rust] return a Result when checking schema compatibility so the end users will have feedback in case or errors
Co-authored-by: Marcos Schroh <[email protected]>

lang/rust/avro/README.md

@@ -634,7 +634,7 @@ use apache_avro::{Schema, schema_compatibility::SchemaCompatibility};
 
 let writers_schema = Schema::parse_str(r#"{"type": "array", "items":"int"}"#).unwrap();
 let readers_schema = Schema::parse_str(r#"{"type": "array", "items":"long"}"#).unwrap();
-assert_eq!(true, SchemaCompatibility::can_read(&writers_schema, &readers_schema));
+assert!(SchemaCompatibility::can_read(&writers_schema, &readers_schema).is_ok());
 ```
 
 2. Incompatible schemas (a long array schema cannot be read by an int array schema)
@@ -647,7 +647,7 @@ use apache_avro::{Schema, schema_compatibility::SchemaCompatibility};
 
 let writers_schema = Schema::parse_str(r#"{"type": "array", "items":"long"}"#).unwrap();
 let readers_schema = Schema::parse_str(r#"{"type": "array", "items":"int"}"#).unwrap();
-assert_eq!(false, SchemaCompatibility::can_read(&writers_schema, &readers_schema));
+assert!(SchemaCompatibility::can_read(&writers_schema, &readers_schema).is_err());
 ```
 
 <!-- cargo-rdme end -->

lang/rust/avro/src/error.rs

@@ -480,6 +480,48 @@ pub enum Error {
     BadCodecMetadata,
 }
 
+#[derive(thiserror::Error, Debug)]
+pub enum CompatibilityError {
+    #[error("Schemas are not compatible. Writer schema is {writer_schema_type}, but reader schema is {reader_schema_type}")]
+    WrongType {
+        writer_schema_type: String,
+        reader_schema_type: String,
+    },
+
+    #[error("Schemas are not compatible. The {schema_type} should have been {expected_type}")]
+    TypeExpected {
+        schema_type: String,
+        expected_type: String,
+    },
+
+    #[error("Schemas are not compatible. Field '{0}' in reader schema does not match the type in the writer schema")]
+    FieldTypeMismatch(String),
+
+    #[error("Schemas are not compatible. Schemas mismatch")]
+    SchemaMismatch,
+
+    #[error("Schemas are not compatible. Field '{0}' in reader schema must have a default value")]
+    MissingDefaultValue(String),
+
+    #[error("Schemas are not compatible. Reader's symbols must contain all writer's symbols")]
+    MissingSymbols,
+
+    #[error("Schemas are not compatible. All elements in union must match for both schemas")]
+    MissingUnionElements,
+
+    #[error("Schemas are not compatible. Name and size don't match for fixed")]
+    FixedMismatch,
+
+    #[error("Schemas are not compatible. The name must be the same for both schemas. Writer's name {writer_name} and reader's name {reader_name}")]
+    NameMismatch {
+        writer_name: String,
+        reader_name: String,
+    },
+
+    #[error("Schemas are not compatible. Unknown type for '{0}'. Make sure that the type is a valid one")]
+    Inconclusive(String),
+}
+
 impl serde::ser::Error for Error {
     fn custom<T: fmt::Display>(msg: T) -> Self {
         Error::SerializeValue(msg.to_string())

lang/rust/avro/src/lib.rs

@@ -747,7 +747,7 @@
 //!
 //! let writers_schema = Schema::parse_str(r#"{"type": "array", "items":"int"}"#).unwrap();
 //! let readers_schema = Schema::parse_str(r#"{"type": "array", "items":"long"}"#).unwrap();
-//! assert_eq!(true, SchemaCompatibility::can_read(&writers_schema, &readers_schema));
+//! assert!(SchemaCompatibility::can_read(&writers_schema, &readers_schema).is_ok());
 //! ```
 //!
 //! 2. Incompatible schemas (a long array schema cannot be read by an int array schema)
@@ -760,7 +760,7 @@
 //!
 //! let writers_schema = Schema::parse_str(r#"{"type": "array", "items":"long"}"#).unwrap();
 //! let readers_schema = Schema::parse_str(r#"{"type": "array", "items":"int"}"#).unwrap();
-//! assert_eq!(false, SchemaCompatibility::can_read(&writers_schema, &readers_schema));
+//! assert!(SchemaCompatibility::can_read(&writers_schema, &readers_schema).is_err());
 //! ```
 
 mod codec;

lang/rust/avro/src/schema.rs

@@ -206,6 +206,42 @@ impl From<&types::Value> for SchemaKind {
     }
 }
 
+// Implement `Display` for `SchemaKind`.
+impl fmt::Display for Schema {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            Schema::Null => write!(f, "Null"),
+            Schema::Boolean => write!(f, "Boolean"),
+            Schema::Int => write!(f, "Int"),
+            Schema::Long => write!(f, "Long"),
+            Schema::Float => write!(f, "Float"),
+            Schema::Double => write!(f, "Double"),
+            Schema::Bytes => write!(f, "Bytes"),
+            Schema::String => write!(f, "String"),
+            Schema::Array(..) => write!(f, "Array"),
+            Schema::Map(..) => write!(f, "Map"),
+            Schema::Union(..) => write!(f, "Union"),
+            Schema::Record(..) => write!(f, "Record"),
+            Schema::Enum(..) => write!(f, "Enum"),
+            Schema::Fixed(..) => write!(f, "Fixed"),
+            Schema::Decimal(..) => write!(f, "Decimal"),
+            Schema::BigDecimal => write!(f, "BigDecimal"),
+            Schema::Uuid => write!(f, "Uuid"),
+            Schema::Date => write!(f, "Date"),
+            Schema::TimeMillis => write!(f, "TimeMillis"),
+            Schema::TimeMicros => write!(f, "TimeMicros"),
+            Schema::TimestampMillis => write!(f, "TimestampMillis"),
+            Schema::TimestampMicros => write!(f, "TimestampMicros"),
+            Schema::LocalTimestampMillis => write!(f, "LocalTimestampMillis"),
+            Schema::LocalTimestampMicros => write!(f, "LocalTimestampMicros"),
+            Schema::Duration => write!(f, "Duration"),
+            Schema::Ref { name } => {
+                write!(f, "{}", name.name)
+            }
+        }
+    }
+}
+
 /// Represents names for `record`, `enum` and `fixed` Avro schemas.
 ///
 /// Each of these `Schema`s have a `fullname` composed of two parts: