AVRO-3799: [Rust] Enable the schema parser to read and parse from inp…

…ut streams for Rust binding (#2352)

* Add a feature reading and parsing schemas from readers.

* Fix format.

* Add Jira number to the test names.

lang/rust/avro/src/error.rs

@@ -259,6 +259,9 @@ pub enum Error {
     #[error("Failed to parse schema from JSON")]
     ParseSchemaJson(#[source] serde_json::Error),
 
+    #[error("Failed to read schema")]
+    ReadSchemaFromReader(#[source] std::io::Error),
+
     #[error("Must be a JSON string, object or array")]
     ParseSchemaFromValidJson,
 

lang/rust/avro/src/schema.rs

@@ -32,6 +32,7 @@ use std::{
     fmt,
     fmt::Debug,
     hash::Hash,
+    io::Read,
     str::FromStr,
 };
 use strum_macros::{EnumDiscriminants, EnumString};
@@ -933,6 +934,15 @@ impl Schema {
         parser.parse_list()
     }
 
+    /// Create a `Schema` from a reader which implements [`Read`].
+    pub fn parse_reader(reader: &mut (impl Read + ?Sized)) -> AvroResult<Schema> {
+        let mut buf = String::new();
+        match reader.read_to_string(&mut buf) {
+            Ok(_) => Self::parse_str(&buf),
+            Err(e) => Err(Error::ReadSchemaFromReader(e)),
+        }
+    }
+
     /// Parses an Avro schema from JSON.
     pub fn parse(value: &Value) -> AvroResult<Schema> {
         let mut parser = Parser::default();

lang/rust/avro/tests/schema.rs

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::io::{Cursor, Read};
+
 use apache_avro::{
     schema::{EnumSchema, FixedSchema, Name, RecordField, RecordSchema},
     to_avro_datum, to_value,
@@ -682,6 +684,61 @@ fn test_parse() -> TestResult {
     Ok(())
 }
 
+#[test]
+fn test_3799_parse_reader() -> TestResult {
+    init();
+    for (raw_schema, valid) in EXAMPLES.iter() {
+        let schema = Schema::parse_reader(&mut Cursor::new(raw_schema));
+        if *valid {
+            assert!(
+                schema.is_ok(),
+                "schema {raw_schema} was supposed to be valid; error: {schema:?}",
+            )
+        } else {
+            assert!(
+                schema.is_err(),
+                "schema {raw_schema} was supposed to be invalid"
+            )
+        }
+    }
+
+    // Ensure it works for trait objects too.
+    for (raw_schema, valid) in EXAMPLES.iter() {
+        let reader: &mut dyn Read = &mut Cursor::new(raw_schema);
+        let schema = Schema::parse_reader(reader);
+        if *valid {
+            assert!(
+                schema.is_ok(),
+                "schema {raw_schema} was supposed to be valid; error: {schema:?}",
+            )
+        } else {
+            assert!(
+                schema.is_err(),
+                "schema {raw_schema} was supposed to be invalid"
+            )
+        }
+    }
+    Ok(())
+}
+
+#[test]
+fn test_3799_raise_io_error_from_parse_read() -> Result<(), String> {
+    // 0xDF is invalid for UTF-8.
+    let mut invalid_data = Cursor::new([0xDF]);
+
+    let error = Schema::parse_reader(&mut invalid_data).unwrap_err();
+
+    if let Error::ReadSchemaFromReader(e) = error {
+        assert!(
+            e.to_string().contains("stream did not contain valid UTF-8"),
+            "{e}"
+        );
+        Ok(())
+    } else {
+        Err(format!("Expected std::io::Error, got {error:?}"))
+    }
+}
+
 #[test]
 /// Test that the string generated by an Avro Schema object is, in fact, a valid Avro schema.
 fn test_valid_cast_to_string_after_parse() -> TestResult {