[AVRO-3506] [rust] Single object writer

lang/csharp/src/apache/main/CodeGen/CodeGen.cs

@@ -1197,7 +1197,7 @@ private static string ReplaceMappedNamespacesInSchema(string input, IEnumerable<
             if (namespaceMapping == null || input == null)
                 return input;
 
-            // Replace namespace in "namespace" definitions: 
+            // Replace namespace in "namespace" definitions:
             //    "namespace": "originalnamespace" -> "namespace": "mappednamespace"
             //    "namespace": "originalnamespace.whatever" -> "namespace": "mappednamespace.whatever"
             // Note: It keeps the original whitespaces

lang/csharp/src/apache/test/Schema/SchemaTests.cs

@@ -172,7 +172,7 @@ private static void testToString(Schema sc, string schema)
         {
             try
             {
-                //remove any excess spaces in the JSON to normalize the match with toString 
+                //remove any excess spaces in the JSON to normalize the match with toString
                 schema = schema.Replace("{ ", "{")
                     .Replace("} ", "}")
                     .Replace("\" ", "\"")
@@ -575,7 +575,7 @@ public void TestUnion(string s, Schema.Type[] types)
 
             UnionSchema schema = UnionSchema.Create(types.Select(t => (Schema)PrimitiveSchema.Create(t)).ToList());
             Assert.AreEqual(sc, schema);
-            
+
             Assert.AreEqual(Schema.Type.Union, sc.Tag);
             UnionSchema us = (UnionSchema)sc;
             Assert.AreEqual(types.Length, us.Count);

lang/csharp/src/apache/test/Util/LogicalTypeTests.cs

@@ -67,7 +67,7 @@ public void TestDecimal(
                 "-123456789123456789.56",
                 "000000000000000001.01",
                 "-000000000000000001.01"
-            )] string s, 
+            )] string s,
             [Values(
                 "\"bytes\"",
                 "{\"type\": \"fixed\", \"size\": 16, \"name\": \"n\"}"
@@ -94,7 +94,7 @@ public void TestDecimalScale(
                 "-1234567891234567890123456789",
                 "0000000000000000000000000001",
                 "-0000000000000000000000000001"
-            )] string s, 
+            )] string s,
             [Values(1, 2, 3, 4, 5, 6, 7, 8)] int scale,
             [Values(
                 "\"bytes\"",
@@ -313,7 +313,7 @@ public void TestLocalTimestampMicrosecond(string s, string e)
             }
 
             expectedDate = expectedDate.ToLocalTime();
-            
+
             var avroLocalTimestampMicro = new LocalTimestampMicrosecond();
             var convertedDate = (DateTime)avroLocalTimestampMicro.ConvertToLogicalValue(avroLocalTimestampMicro.ConvertToBaseValue(date, schema), schema);
             Assert.AreEqual(expectedDate, convertedDate);
@@ -342,7 +342,7 @@ public void TestTimeMillisecond(string s, string e, bool expectRangeError)
             var timeMilliSchema = (LogicalSchema)Schema.Parse("{\"type\": \"int\", \"logicalType\": \"time-millis\"}");
 
             var time = TimeSpan.Parse(s);
-            
+
             var avroTimeMilli = new TimeMillisecond();
 
             if (expectRangeError)
@@ -385,7 +385,7 @@ public void TestTimeMicrosecond(string s, string e, bool expectRangeError)
             var timeMicroSchema = (LogicalSchema)Schema.Parse("{\"type\": \"long\", \"logicalType\": \"time-micros\"}");
 
             var time = TimeSpan.Parse(s);
-            
+
             var avroTimeMicro = new TimeMicrosecond();
 
             if (expectRangeError)

lang/java/avro/src/test/java/org/apache/avro/message/TestGenerateInteropSingleObjectEncoding.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.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/**
+ * Generates <code>test_message.bin</code> - a <a href=
+ * "https://avro.apache.org/docs/current/spec.html#single_object_encoding">single
+ * object encoded</a> Avro message.
+ */
+public class TestGenerateInteropSingleObjectEncoding {
+  private static final String RESOURCES_FOLDER = System.getProperty("share.dir", "../../../share")
+      + "/test/data/messageV1";
+  private static final File SCHEMA_FILE = new File(RESOURCES_FOLDER + "/test_schema.avsc");
+  private static final File MESSAGE_FILE = new File(RESOURCES_FOLDER + "/test_message.bin");
+  private static Schema SCHEMA;
+  private static GenericRecordBuilder BUILDER;
+
+  @BeforeClass
+  public static void setup() throws IOException {
+    try (FileInputStream fileInputStream = new FileInputStream(SCHEMA_FILE)) {
+      SCHEMA = new Schema.Parser().parse(fileInputStream);
+      BUILDER = new GenericRecordBuilder(SCHEMA);
+    }
+  }
+
+  @Test
+  public void generateData() throws IOException {
+    MessageEncoder<GenericData.Record> encoder = new BinaryMessageEncoder<>(GenericData.get(), SCHEMA);
+    BUILDER.set("id", 42L).set("name", "Bill").set("tags", Arrays.asList("dog_lover", "cat_hater")).build();
+    ByteBuffer buffer = encoder.encode(
+        BUILDER.set("id", 42L).set("name", "Bill").set("tags", Arrays.asList("dog_lover", "cat_hater")).build());
+    new FileOutputStream(MESSAGE_FILE).write(buffer.array());
+  }
+}

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

@@ -0,0 +1,65 @@
+/*
+ * 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.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.Arrays;
+
+/**
+ * Tests that <code>test_message.bin</code> is properly encoded <a href=
+ * "https://avro.apache.org/docs/current/spec.html#single_object_encoding">single
+ * object</a>
+ */
+public class TestInteropSingleObjectEncoding {
+  private static final String RESOURCES_FOLDER = System.getProperty("share.dir", "../../../share")
+      + "/test/data/messageV1";
+  private static final File SCHEMA_FILE = new File(RESOURCES_FOLDER + "/test_schema.avsc");
+  private static final File MESSAGE_FILE = new File(RESOURCES_FOLDER + "/test_message.bin");
+  private static Schema SCHEMA;
+  private static GenericRecordBuilder BUILDER;
+
+  @BeforeClass
+  public static void setup() throws IOException {
+    try (FileInputStream fileInputStream = new FileInputStream(SCHEMA_FILE)) {
+      SCHEMA = new Schema.Parser().parse(fileInputStream);
+      BUILDER = new GenericRecordBuilder(SCHEMA);
+    }
+  }
+
+  @Test
+  public void checkSingleObjectEncoding() 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_single_object_encoding.rs

@@ -0,0 +1,60 @@
+// 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};
+
+const RESOURCES_FOLDER: &str = "../../share/test/data/messageV1";
+
+struct InteropMessage;
+
+impl AvroSchema for InteropMessage {
+    fn get_schema() -> apache_avro::Schema {
+        let schema = std::fs::read_to_string(format!("{}/test_schema.avsc", RESOURCES_FOLDER))
+            .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(format!("{}/test_message.bin", RESOURCES_FOLDER))
+        .expect("File with single object not found or error occurred while 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

@@ -411,6 +411,10 @@ pub enum Error {
         value_kind: ValueKind,
         supported_schema: Vec<SchemaKind>,
     },
+    #[error(
+        "Internal buffer not drained properly. Re-initialize the single object 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;