You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@avro.apache.org by mg...@apache.org on 2022/05/09 11:52:09 UTC
[avro] branch branch-1.11 updated: [AVRO-3507] Implement Single Object reader (#1681)
This is an automated email from the ASF dual-hosted git repository.
mgrigorov pushed a commit to branch branch-1.11
in repository https://gitbox.apache.org/repos/asf/avro.git
The following commit(s) were added to refs/heads/branch-1.11 by this push:
new 2f1746490 [AVRO-3507] Implement Single Object reader (#1681)
2f1746490 is described below
commit 2f17464908194b7c6b76bd04cca20b107ed4d075
Author: Jack Klamer <jf...@gmail.com>
AuthorDate: Mon May 9 06:51:40 2022 -0500
[AVRO-3507] Implement Single Object reader (#1681)
* Readers
* unit tested
* interop tested
* Update lang/rust/avro/src/reader.rs
Co-authored-by: Martin Grigorov <ma...@users.noreply.github.com>
* Apply suggestions from code review
Co-authored-by: Martin Grigorov <ma...@users.noreply.github.com>
* one read for interop+ error message
* AVRO-3507: Improve an error message in a unit test
Signed-off-by: Martin Tzvetanov Grigorov <mg...@apache.org>
* AVRO-3507: Better variable names and error messages
Signed-off-by: Martin Tzvetanov Grigorov <mg...@apache.org>
* AVRO-3507: Fix Clippy 1.51 errors
Signed-off-by: Martin Tzvetanov Grigorov <mg...@apache.org>
Co-authored-by: Martin Grigorov <ma...@users.noreply.github.com>
Co-authored-by: Martin Tzvetanov Grigorov <mg...@apache.org>
(cherry picked from commit 41fbaa8e9443bebc276dc8f2d056a155f47bf0c8)
---
.../test_interop_single_object_encoding.rs | 29 ++-
lang/rust/avro/src/decode.rs | 14 +-
lang/rust/avro/src/error.rs | 3 +
lang/rust/avro/src/lib.rs | 2 +-
lang/rust/avro/src/reader.rs | 255 ++++++++++++++++++++-
lang/rust/build.sh | 2 +
6 files changed, 291 insertions(+), 14 deletions(-)
diff --git a/lang/rust/avro/examples/test_interop_single_object_encoding.rs b/lang/rust/avro/examples/test_interop_single_object_encoding.rs
index f03e938cc..06fb753ff 100644
--- a/lang/rust/avro/examples/test_interop_single_object_encoding.rs
+++ b/lang/rust/avro/examples/test_interop_single_object_encoding.rs
@@ -49,12 +49,27 @@ impl From<InteropMessage> for Value {
}
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();
+ let single_object = std::fs::read(format!("{}/test_message.bin", RESOURCES_FOLDER))
+ .expect("File with single object not found or error occurred while reading it.");
+ test_write(&single_object);
+ test_read(single_object);
+}
+
+fn test_write(expected: &[u8]) {
+ let mut encoded: Vec<u8> = Vec::new();
apache_avro::SpecificSingleObjectWriter::<InteropMessage>::with_capacity(1024)
- .expect("resolve expected")
- .write_value(InteropMessage, &mut generated_encoding)
- .expect("Should encode");
- assert_eq!(file_message, generated_encoding)
+ .expect("Resolving failed")
+ .write_value(InteropMessage, &mut encoded)
+ .expect("Encoding failed");
+ assert_eq!(expected, &encoded)
+}
+
+fn test_read(encoded: Vec<u8>) {
+ let mut encoded = &encoded[..];
+ let read_message = apache_avro::GenericSingleObjectReader::new(InteropMessage::get_schema())
+ .expect("Resolving failed")
+ .read_value(&mut encoded)
+ .expect("Decoding failed");
+ let expected_value: Value = InteropMessage.into();
+ assert_eq!(expected_value, read_message)
}
diff --git a/lang/rust/avro/src/decode.rs b/lang/rust/avro/src/decode.rs
index 7ddb9fc11..d88157307 100644
--- a/lang/rust/avro/src/decode.rs
+++ b/lang/rust/avro/src/decode.rs
@@ -18,12 +18,13 @@
use crate::{
decimal::Decimal,
duration::Duration,
- schema::{NamesRef, Namespace, ResolvedSchema, Schema},
+ schema::{Name, Namespace, ResolvedSchema, Schema},
types::Value,
util::{safe_len, zag_i32, zag_i64},
AvroResult, Error,
};
use std::{
+ borrow::Borrow,
collections::HashMap,
convert::TryFrom,
io::{ErrorKind, Read},
@@ -72,9 +73,9 @@ pub fn decode<R: Read>(schema: &Schema, reader: &mut R) -> AvroResult<Value> {
decode_internal(schema, rs.get_names(), &None, reader)
}
-fn decode_internal<R: Read>(
+pub(crate) fn decode_internal<R: Read, S: Borrow<Schema>>(
schema: &Schema,
- names: &NamesRef,
+ names: &HashMap<Name, S>,
enclosing_namespace: &Namespace,
reader: &mut R,
) -> AvroResult<Value> {
@@ -273,7 +274,12 @@ fn decode_internal<R: Read>(
Schema::Ref { ref name } => {
let fully_qualified_name = name.fully_qualified_name(enclosing_namespace);
if let Some(resolved) = names.get(&fully_qualified_name) {
- decode_internal(resolved, names, &fully_qualified_name.namespace, reader)
+ decode_internal(
+ resolved.borrow(),
+ names,
+ &fully_qualified_name.namespace,
+ reader,
+ )
} else {
Err(Error::SchemaResolutionError(fully_qualified_name))
}
diff --git a/lang/rust/avro/src/error.rs b/lang/rust/avro/src/error.rs
index d5427e9af..1e1e685ab 100644
--- a/lang/rust/avro/src/error.rs
+++ b/lang/rust/avro/src/error.rs
@@ -343,6 +343,9 @@ pub enum Error {
#[error("wrong magic in header")]
HeaderMagic,
+ #[error("Message Header mismatch. Expected: {0:?}. Actual: {1:?}")]
+ SingleObjectHeaderMismatch([u8; 10], [u8; 10]),
+
#[error("Failed to get JSON from avro.schema key in map")]
GetAvroSchemaFromMap,
diff --git a/lang/rust/avro/src/lib.rs b/lang/rust/avro/src/lib.rs
index 4d713a7f9..104f9a5d7 100644
--- a/lang/rust/avro/src/lib.rs
+++ b/lang/rust/avro/src/lib.rs
@@ -742,7 +742,7 @@ pub use de::from_value;
pub use decimal::Decimal;
pub use duration::{Days, Duration, Millis, Months};
pub use error::Error;
-pub use reader::{from_avro_datum, Reader};
+pub use reader::{from_avro_datum, GenericSingleObjectReader, Reader, SpecificSingleObjectReader};
pub use schema::Schema;
pub use ser::to_value;
pub use util::max_allocation_bytes;
diff --git a/lang/rust/avro/src/reader.rs b/lang/rust/avro/src/reader.rs
index 0b75e6461..f2e0f38b6 100644
--- a/lang/rust/avro/src/reader.rs
+++ b/lang/rust/avro/src/reader.rs
@@ -16,11 +16,21 @@
// under the License.
//! Logic handling reading from Avro format at user level.
-use crate::{decode::decode, schema::Schema, types::Value, util, AvroResult, Codec, Error};
+use crate::{
+ decode::{decode, decode_internal},
+ from_value,
+ rabin::Rabin,
+ schema::{AvroSchema, ResolvedOwnedSchema, Schema},
+ types::Value,
+ util, AvroResult, Codec, Error,
+};
+use serde::de::DeserializeOwned;
use serde_json::from_slice;
use std::{
collections::HashMap,
+ convert::TryFrom,
io::{ErrorKind, Read},
+ marker::PhantomData,
str::FromStr,
};
@@ -343,10 +353,97 @@ pub fn from_avro_datum<R: Read>(
}
}
+pub struct GenericSingleObjectReader {
+ write_schema: ResolvedOwnedSchema,
+ expected_header: [u8; 10],
+}
+
+impl GenericSingleObjectReader {
+ pub fn new(schema: Schema) -> AvroResult<GenericSingleObjectReader> {
+ let fingerprint = schema.fingerprint::<Rabin>();
+ let expected_header = [
+ 0xC3,
+ 0x01,
+ fingerprint.bytes[0],
+ fingerprint.bytes[1],
+ fingerprint.bytes[2],
+ fingerprint.bytes[3],
+ fingerprint.bytes[4],
+ fingerprint.bytes[5],
+ fingerprint.bytes[6],
+ fingerprint.bytes[7],
+ ];
+ Ok(GenericSingleObjectReader {
+ write_schema: ResolvedOwnedSchema::try_from(schema)?,
+ expected_header,
+ })
+ }
+
+ pub fn read_value<R: Read>(&self, reader: &mut R) -> AvroResult<Value> {
+ let mut header: [u8; 10] = [0; 10];
+ match reader.read(&mut header) {
+ Ok(size) => {
+ if size == 10 && self.expected_header == header {
+ decode_internal(
+ self.write_schema.get_root_schema(),
+ self.write_schema.get_names(),
+ &None,
+ reader,
+ )
+ } else {
+ Err(Error::SingleObjectHeaderMismatch(
+ self.expected_header,
+ header,
+ ))
+ }
+ }
+ Err(io_error) => Err(Error::ReadHeader(io_error)),
+ }
+ }
+}
+
+pub struct SpecificSingleObjectReader<T>
+where
+ T: AvroSchema,
+{
+ inner: GenericSingleObjectReader,
+ _model: PhantomData<T>,
+}
+
+impl<T> SpecificSingleObjectReader<T>
+where
+ T: AvroSchema,
+{
+ pub fn new() -> AvroResult<SpecificSingleObjectReader<T>> {
+ Ok(SpecificSingleObjectReader {
+ inner: GenericSingleObjectReader::new(T::get_schema())?,
+ _model: PhantomData,
+ })
+ }
+}
+
+impl<T> SpecificSingleObjectReader<T>
+where
+ T: AvroSchema + From<Value>,
+{
+ pub fn read_from_value<R: Read>(&self, reader: &mut R) -> AvroResult<T> {
+ self.inner.read_value(reader).map(|v| v.into())
+ }
+}
+
+impl<T> SpecificSingleObjectReader<T>
+where
+ T: AvroSchema + DeserializeOwned,
+{
+ pub fn read<R: Read>(&self, reader: &mut R) -> AvroResult<T> {
+ from_value::<T>(&self.inner.read_value(reader)?)
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
- use crate::{from_value, types::Record, Reader};
+ use crate::{encode::encode, from_value, types::Record, Reader};
use serde::Deserialize;
use std::io::Cursor;
@@ -569,4 +666,158 @@ mod tests {
let reader = Reader::new(&result[..]).unwrap();
assert_eq!(reader.user_metadata(), &user_meta_data);
}
+
+ #[derive(Deserialize, Clone, PartialEq, Debug)]
+ struct TestSingleObjectReader {
+ a: i64,
+ b: f64,
+ c: Vec<String>,
+ }
+
+ impl AvroSchema for TestSingleObjectReader {
+ fn get_schema() -> Schema {
+ let schema = r#"
+ {
+ "type":"record",
+ "name":"TestSingleObjectWrtierSerialize",
+ "fields":[
+ {
+ "name":"a",
+ "type":"long"
+ },
+ {
+ "name":"b",
+ "type":"double"
+ },
+ {
+ "name":"c",
+ "type":{
+ "type":"array",
+ "items":"string"
+ }
+ }
+ ]
+ }
+ "#;
+ Schema::parse_str(schema).unwrap()
+ }
+ }
+
+ impl From<Value> for TestSingleObjectReader {
+ fn from(obj: Value) -> TestSingleObjectReader {
+ if let Value::Record(fields) = obj {
+ let mut a = None;
+ let mut b = None;
+ let mut c = vec![];
+ for (field_name, v) in fields {
+ match (field_name.as_str(), v) {
+ ("a", Value::Long(i)) => a = Some(i),
+ ("b", Value::Double(d)) => b = Some(d),
+ ("c", Value::Array(v)) => {
+ for inner_val in v {
+ if let Value::String(s) = inner_val {
+ c.push(s);
+ }
+ }
+ }
+ (key, value) => panic!("Unexpected pair: {:?} -> {:?}", key, value),
+ }
+ }
+ TestSingleObjectReader {
+ a: a.unwrap(),
+ b: b.unwrap(),
+ c,
+ }
+ } else {
+ panic!("Expected a Value::Record but was {:?}", obj)
+ }
+ }
+ }
+
+ impl From<TestSingleObjectReader> for Value {
+ fn from(obj: TestSingleObjectReader) -> Value {
+ Value::Record(vec![
+ ("a".into(), obj.a.into()),
+ ("b".into(), obj.b.into()),
+ (
+ "c".into(),
+ Value::Array(obj.c.into_iter().map(|s| s.into()).collect()),
+ ),
+ ])
+ }
+ }
+
+ #[test]
+ fn test_avro_3507_single_object_reader() {
+ let obj = TestSingleObjectReader {
+ a: 42,
+ b: 3.33,
+ c: vec!["cat".into(), "dog".into()],
+ };
+ let mut to_read = Vec::<u8>::new();
+ to_read.extend_from_slice(&[0xC3, 0x01]);
+ to_read.extend_from_slice(
+ &TestSingleObjectReader::get_schema()
+ .fingerprint::<Rabin>()
+ .bytes[..],
+ );
+ encode(
+ &obj.clone().into(),
+ &TestSingleObjectReader::get_schema(),
+ &mut to_read,
+ )
+ .expect("Encode should succeed");
+ let mut to_read = &to_read[..];
+ let generic_reader = GenericSingleObjectReader::new(TestSingleObjectReader::get_schema())
+ .expect("Schema should resolve");
+ let val = generic_reader
+ .read_value(&mut to_read)
+ .expect("Should read");
+ let expected_value: Value = obj.into();
+ assert_eq!(expected_value, val);
+ }
+
+ #[test]
+ fn test_avro_3507_reader_parity() {
+ let obj = TestSingleObjectReader {
+ a: 42,
+ b: 3.33,
+ c: vec!["cat".into(), "dog".into()],
+ };
+
+ let mut to_read = Vec::<u8>::new();
+ to_read.extend_from_slice(&[0xC3, 0x01]);
+ to_read.extend_from_slice(
+ &TestSingleObjectReader::get_schema()
+ .fingerprint::<Rabin>()
+ .bytes[..],
+ );
+ encode(
+ &obj.clone().into(),
+ &TestSingleObjectReader::get_schema(),
+ &mut to_read,
+ )
+ .expect("Encode should succeed");
+ let generic_reader = GenericSingleObjectReader::new(TestSingleObjectReader::get_schema())
+ .expect("Schema should resolve");
+ let specific_reader = SpecificSingleObjectReader::<TestSingleObjectReader>::new()
+ .expect("schema should resolve");
+ let mut to_read1 = &to_read[..];
+ let mut to_read2 = &to_read[..];
+ let mut to_read3 = &to_read[..];
+
+ let val = generic_reader
+ .read_value(&mut to_read1)
+ .expect("Should read");
+ let read_obj1 = specific_reader
+ .read_from_value(&mut to_read2)
+ .expect("Should read from value");
+ let read_obj2 = specific_reader
+ .read(&mut to_read3)
+ .expect("Should read from deserilize");
+ let expected_value: Value = obj.clone().into();
+ assert_eq!(obj, read_obj1);
+ assert_eq!(obj, read_obj2);
+ assert_eq!(val, expected_value)
+ }
}
diff --git a/lang/rust/build.sh b/lang/rust/build.sh
index c44c05ad2..c7caffbc5 100755
--- a/lang/rust/build.sh
+++ b/lang/rust/build.sh
@@ -63,7 +63,9 @@ do
interop-data-test)
prepare_build
+ echo "Running interop data tests"
cargo run --all-features --example test_interop_data
+ echo -e "\nRunning single object encoding interop data tests"
cargo run --all-features --example test_interop_single_object_encoding
;;
*)