[AVRO-3506] [rust] Single object writer

lang/java/avro/src/test/java/org/apache/avro/message/TestInteropMessageData.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.avro.message;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.logging.Logger;
+
+public class TestInteropMessageData {
+  private static final String inDir = System.getProperty("share.dir", "../../../share") + "/test/data/messageV1";
+  private static final File SCHEMA_FILE = new File(inDir + "/test_schema.avsc");
+  private static final File MESSAGE_FILE = new File(inDir + "/test_message.bin");
+  private static Schema SCHEMA;
+  private static GenericRecordBuilder BUILDER;
+
+  @BeforeClass
+  public static void setup() throws IOException {
+    final FileInputStream fileInputStream = new FileInputStream(SCHEMA_FILE);
+    SCHEMA = new Schema.Parser().parse(fileInputStream);
+    BUILDER = new GenericRecordBuilder(SCHEMA);
+    fileInputStream.close();
+  }
+
+  @Test
+  public void sanity_check() throws IOException {
+    MessageEncoder<GenericData.Record> encoder = new BinaryMessageEncoder<>(GenericData.get(), SCHEMA);
+    ByteBuffer buffer = encoder.encode(
+        BUILDER.set("id", 42L).set("name", "Bill").set("tags", Arrays.asList("dog_lover", "cat_hater")).build());
+    byte[] fileBuffer = Files.readAllBytes(MESSAGE_FILE.toPath());
+    Assert.assertArrayEquals(fileBuffer, buffer.array());
+  }
+}

lang/rust/avro/examples/test_interop_message_data.rs

@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use apache_avro::{schema::AvroSchema, types::Value};
+
+struct InteropMessage;
+
+impl AvroSchema for InteropMessage {
+    fn get_schema() -> apache_avro::Schema {
+        let schema = std::fs::read_to_string("../../share/test/data/messageV1/test_schema.avsc")
+            .expect("File should exist with schema inside");
+        apache_avro::Schema::parse_str(schema.as_str())
+            .expect("File should exist with schema inside")
+    }
+}
+
+impl From<InteropMessage> for Value {
+    fn from(_: InteropMessage) -> Value {
+        Value::Record(vec![
+            ("id".into(), 42i64.into()),
+            ("name".into(), "Bill".into()),
+            (
+                "tags".into(),
+                Value::Array(
+                    vec!["dog_lover", "cat_hater"]
+                        .into_iter()
+                        .map(|s| s.into())
+                        .collect(),
+                ),
+            ),
+        ])
+    }
+}
+
+fn main() {
+    let file_message = std::fs::read("../../share/test/data/messageV1/test_message.bin")
+        .expect("File not found or error reading");
+    let mut generated_encoding: Vec<u8> = Vec::new();
+    apache_avro::SingleObjectWriter::<InteropMessage>::with_capacity(1024)
+        .expect("resolve expected")
+        .write_value(InteropMessage, &mut generated_encoding)
+        .expect("Should encode");
+    assert_eq!(file_message, generated_encoding)
+}

lang/rust/avro/src/encode.rs

@@ -16,12 +16,16 @@
 // under the License.
 
 use crate::{
-    schema::{NamesRef, Namespace, ResolvedSchema, Schema, SchemaKind},
+    schema::{Name, Namespace, ResolvedSchema, Schema, SchemaKind},
     types::{Value, ValueKind},
     util::{zig_i32, zig_i64},
     AvroResult, Error,
 };
-use std::convert::{TryFrom, TryInto};
+use std::{
+    borrow::Borrow,
+    collections::HashMap,
+    convert::{TryFrom, TryInto},
+};
 
 /// Encode a `Value` into avro format.
 ///
@@ -47,19 +51,19 @@ fn encode_int(i: i32, buffer: &mut Vec<u8>) {
     zig_i32(i, buffer)
 }
 
-pub(crate) fn encode_internal(
+pub(crate) fn encode_internal<S: Borrow<Schema>>(
     value: &Value,
     schema: &Schema,
-    names: &NamesRef,
+    names: &HashMap<Name, S>,
     enclosing_namespace: &Namespace,
     buffer: &mut Vec<u8>,
 ) -> AvroResult<()> {
     if let Schema::Ref { ref name } = schema {
         let fully_qualified_name = name.fully_qualified_name(enclosing_namespace);
-        let resolved = *names
+        let resolved = names
             .get(&fully_qualified_name)
             .ok_or(Error::SchemaResolutionError(fully_qualified_name))?;
-        return encode_internal(value, resolved, names, enclosing_namespace, buffer);
+        return encode_internal(value, resolved.borrow(), names, enclosing_namespace, buffer);
     }
 
     match value {

lang/rust/avro/src/error.rs

@@ -408,6 +408,8 @@ pub enum Error {
         value_kind: ValueKind,
         supported_schema: Vec<SchemaKind>,
     },
+    #[error("Internal buffer not drained properly. Reinitialize the writer struct")]
+    IllegalSingleObjectWriterState,
 }
 
 impl serde::ser::Error for Error {

lang/rust/avro/src/lib.rs

@@ -746,7 +746,7 @@ pub use reader::{from_avro_datum, Reader};
 pub use schema::Schema;
 pub use ser::to_value;
 pub use util::max_allocation_bytes;
-pub use writer::{to_avro_datum, Writer};
+pub use writer::{to_avro_datum, GenericSingleObjectWriter, SingleObjectWriter, Writer};
 
 #[macro_use]
 extern crate log;

lang/rust/avro/src/schema.rs

@@ -434,6 +434,86 @@ impl<'s> ResolvedSchema<'s> {
     }
 }
 
+pub(crate) struct ResolvedOwnedSchema {
+    names: Names,
+    root_schema: Schema,
+}
+
+impl<'s> TryFrom<Schema> for ResolvedOwnedSchema {
+    type Error = Error;
+
+    fn try_from(schema: Schema) -> AvroResult<Self> {
+        let names = HashMap::new();
+        let mut rs = ResolvedOwnedSchema {
+            names,
+            root_schema: schema,
+        };
+        Self::from_internal(&rs.root_schema, &mut rs.names, &None)?;
+        Ok(rs)
+    }
+}
+
+impl ResolvedOwnedSchema {
+    pub(crate) fn get_root_schema(&self) -> &Schema {
+        &self.root_schema
+    }
+    pub(crate) fn get_names(&self) -> &Names {
+        &self.names
+    }
+
+    fn from_internal(
+        schema: &Schema,
+        names: &mut Names,
+        enclosing_namespace: &Namespace,
+    ) -> AvroResult<()> {
+        match schema {
+            Schema::Array(schema) | Schema::Map(schema) => {
+                Self::from_internal(schema, names, enclosing_namespace)
+            }
+            Schema::Union(UnionSchema { schemas, .. }) => {
+                for schema in schemas {
+                    Self::from_internal(schema, names, enclosing_namespace)?
+                }
+                Ok(())
+            }
+            Schema::Enum { name, .. } | Schema::Fixed { name, .. } => {
+                let fully_qualified_name = name.fully_qualified_name(enclosing_namespace);
+                if names
+                    .insert(fully_qualified_name.clone(), schema.clone())
+                    .is_some()
+                {
+                    Err(Error::AmbiguousSchemaDefinition(fully_qualified_name))
+                } else {
+                    Ok(())
+                }
+            }
+            Schema::Record { name, fields, .. } => {
+                let fully_qualified_name = name.fully_qualified_name(enclosing_namespace);
+                if names
+                    .insert(fully_qualified_name.clone(), schema.clone())
+                    .is_some()
+                {
+                    Err(Error::AmbiguousSchemaDefinition(fully_qualified_name))
+                } else {
+                    let record_namespace = fully_qualified_name.namespace;
+                    for field in fields {
+                        Self::from_internal(&field.schema, names, &record_namespace)?
+                    }
+                    Ok(())
+                }
+            }
+            Schema::Ref { name } => {
+                let fully_qualified_name = name.fully_qualified_name(enclosing_namespace);
+                names
+                    .get(&fully_qualified_name)
+                    .map(|_| ())
+                    .ok_or(Error::SchemaResolutionError(fully_qualified_name))
+            }
+            _ => Ok(()),
+        }
+    }
+}
+
 /// Represents a `field` in a `record` Avro schema.
 #[derive(Clone, Debug, PartialEq)]
 pub struct RecordField {

lang/rust/avro/src/types.rs

@@ -20,7 +20,8 @@ use crate::{
     decimal::Decimal,
     duration::Duration,
     schema::{
-        NamesRef, Precision, RecordField, ResolvedSchema, Scale, Schema, SchemaKind, UnionSchema,
+        Name, NamesRef, Precision, RecordField, ResolvedSchema, Scale, Schema, SchemaKind,
+        UnionSchema,
     },
     AvroResult, Error,
 };
@@ -355,7 +356,11 @@ impl Value {
         }
     }
 
-    pub(crate) fn validate_internal(&self, schema: &Schema, names: &NamesRef) -> Option<String> {
+    pub(crate) fn validate_internal<S: std::borrow::Borrow<Schema>>(
+        &self,
+        schema: &Schema,
+        names: &HashMap<Name, S>,
+    ) -> Option<String> {
         match (self, schema) {
             (_, &Schema::Ref { ref name }) => names.get(name).map_or_else(
                 || {
@@ -365,7 +370,7 @@ impl Value {
                         names.keys()
                     ));
                 },
-                |s| self.validate_internal(s, names),
+                |s| self.validate_internal(s.borrow(), names),
             ),
             (&Value::Null, &Schema::Null) => None,
             (&Value::Boolean(_), &Schema::Boolean) => None,
@@ -1080,7 +1085,7 @@ mod tests {
         ];
 
         for (value, schema, valid, expected_err_message) in value_schema_valid.into_iter() {
-            let err_message = value.validate_internal(&schema, &HashMap::default());
+            let err_message = value.validate_internal::<Schema>(&schema, &HashMap::default());
             assert_eq!(valid, err_message.is_none());
             if !valid {
                 let full_err_message = format!(