Hello
I've written some failing rust unit tests showing:
- Non trivial union schemas (without null variant) failing to encode/decode
round-trip (serde)
- Non trivial union schemas including null variant at position 0 failing to
round trip (serde)
- Non trivial union schemas including null varant at position other than 0
failing to round trip (serde)
I've also attached and appended an update to the encoder that makes the
above tests pass. Is this a patch that the team is interested in?
Thanks
Simon
diff --git a/lang/rust/avro/src/encode.rs b/lang/rust/avro/src/encode.rs
index 4593779ac..829a8ee6c 100644
--- a/lang/rust/avro/src/encode.rs
+++ b/lang/rust/avro/src/encode.rs
@@ -19,7 +19,7 @@ use crate::{
decimal::serialize_big_decimal,
schema::{
DecimalSchema, EnumSchema, FixedSchema, Name, Namespace,
RecordSchema, ResolvedSchema,
- Schema, SchemaKind,
+ UnionSchema, Schema, SchemaKind,
},
types::{Value, ValueKind},
util::{zig_i32, zig_i64},
@@ -71,7 +71,20 @@ pub(crate) fn encode_internal<S: Borrow<Schema>>(
}
match value {
- Value::Null => (),
+ Value::Null => {
+ match schema {
+ Schema::Union(s) => {
+ match s.schemas.iter().position(|sch|*sch ==
Schema::Null) {
+ None =>
+ return Err(Error::EncodeValueAsSchemaError {
+ value_kind: ValueKind::Null,
+ supported_schema: vec![SchemaKind::Null,
SchemaKind::Union], }),
+ Some(p) => encode_long(p as i64, buffer),
+ }
+ }
+ _ => ()
+ }
+ },
Value::Boolean(b) => buffer.push(u8::from(*b)),
// Pattern | Pattern here to signify that these _must_ have the
same encoding.
Value::Int(i) | Value::Date(i) | Value::TimeMillis(i) =>
encode_int(*i, buffer),
@@ -242,6 +255,21 @@ pub(crate) fn encode_internal<S: Borrow<Schema>>(
));
}
}
+ } else if let Schema::Union(UnionSchema{ schemas, .. }) =
schema {
+ let original_size = buffer.len();
+ for (index,s) in schemas.iter().enumerate() {
+ encode_long(index as i64, buffer);
+ match encode_internal(value, s.borrow(), names,
enclosing_namespace, buffer) {
+ Ok(_) => return Ok(()),
+ Err(e) => {
+ buffer.truncate(original_size); //undo any
partial encoding
+ }
+ }
+ }
+ return Err(Error::EncodeValueAsSchemaError {
+ value_kind: ValueKind::Record,
+ supported_schema: vec![SchemaKind::Record,
SchemaKind::Union],
+ });
} else {
error!("invalid schema type for Record: {:?}", schema);
return Err(Error::EncodeValueAsSchemaError {
diff --git a/lang/rust/avro/tests/union_schema.rs b/lang/rust/avro/tests/
union_schema.rs
index e69de29bb..1dc19d25c 100644
--- a/lang/rust/avro/tests/union_schema.rs
+++ b/lang/rust/avro/tests/union_schema.rs
@@ -0,0 +1,185 @@
+// 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 serde::{Deserialize, Serialize};
+use serde::de::DeserializeOwned;
+use apache_avro::{from_avro_datum, to_avro_datum, to_value, from_value,
types, Schema, Writer, Reader, Codec};
+
+
+static SCHEMA_A_STR: &str = r#"{
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {"name": "field_a", "type": "float"}
+ ]
+ }"#;
+
+static SCHEMA_B_STR: &str = r#"{
+ "name": "B",
+ "type": "record",
+ "fields": [
+ {"name": "field_b", "type": "long"}
+ ]
+ }"#;
+
+static SCHEMA_C_STR: &str = r#"{
+ "name": "C",
+ "type": "record",
+ "fields": [
+ {"name": "field_union", "type": ["A", "B"]},
+ {"name": "field_c", "type": "string"}
+ ]
+ }"#;
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+struct A {
+ field_a: f32,
+}
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+struct B {
+ field_b: i64,
+}
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+#[serde(untagged)]
+enum UnionAB {
+ A(A),
+ B(B),
+}
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+struct C {
+ field_union: UnionAB,
+ field_c: String
+}
+
+fn encode_decode<T> (input: &T,schema: &Schema,schemata: &Vec<Schema>) -> T
+ where T: DeserializeOwned + Serialize {
+ let mut encoded: Vec<u8> = Vec::new();
+ let mut writer = Writer::with_schemata(&schema,
schemata.iter().collect(), &mut encoded, Codec::Null);
+ writer.append_ser((input.clone())).unwrap();
+ writer.flush().unwrap();
+
+ let mut reader = Reader::with_schemata(schema,
schemata.iter().collect(), encoded.as_slice()).unwrap();
+ from_value::<T>(&reader.next().unwrap().unwrap()).unwrap()
+}
+
+
+#[test]
+fn union_schema_round_trip_no_null() {
+ let schemata: Vec<Schema> = Schema::parse_list(&[SCHEMA_A_STR,
SCHEMA_B_STR, SCHEMA_C_STR]).expect("parsing schemata");
+
+ {
+ let input = C { field_union: (UnionAB::A(A { field_a: 45.5 })),
field_c: "foo".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+ {
+ let input = C { field_union: (UnionAB::B(B { field_b: 73 })),
field_c: "bar".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+}
+
+static SCHEMA_D_STR: &str = r#"{
+ "name": "D",
+ "type": "record",
+ "fields": [
+ {"name": "field_union", "type": ["null", "A", "B"]},
+ {"name": "field_d", "type": "string"}
+ ]
+ }"#;
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+#[serde(untagged)]
+enum UnionNoneAB {
+ None,
+ A(A),
+ B(B),
+}
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+struct D {
+ field_union: UnionNoneAB,
+ field_d: String
+}
+
+#[test]
+fn union_schema_round_trip_null_at_start() {
+ let schemata: Vec<Schema> = Schema::parse_list(&[SCHEMA_A_STR,
SCHEMA_B_STR, SCHEMA_D_STR]).expect("parsing schemata");
+
+ {
+ let input = D { field_union: UnionNoneAB::A(A { field_a: 54.25 }),
field_d: "fooy".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+ {
+ let input = D { field_union: UnionNoneAB::None, field_d:
"fooyy".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+ {
+ let input = D { field_union: UnionNoneAB::B(B { field_b: 103 }),
field_d: "foov".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+}
+
+static SCHEMA_E_STR: &str = r#"{
+ "name": "E",
+ "type": "record",
+ "fields": [
+ {"name": "field_union", "type": ["A", "null", "B"]},
+ {"name": "field_e", "type": "string"}
+ ]
+ }"#;
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+#[serde(untagged)]
+enum UnionANoneB {
+ A(A),
+ None,
+ B(B),
+}
+
+#[derive(Serialize,Deserialize,Clone, PartialEq, Debug)]
+struct E {
+ field_union: UnionANoneB,
+ field_e: String
+}
+
+#[test]
+fn union_schema_round_trip_with_out_of_order_null() {
+ let schemata: Vec<Schema> = Schema::parse_list(&[SCHEMA_A_STR,
SCHEMA_B_STR, SCHEMA_E_STR]).expect("parsing schemata");
+
+ {
+ let input = E { field_union: UnionANoneB::A(A { field_a: 23.75 }),
field_e: "barme".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+ {
+ let input = E { field_union: UnionANoneB::None, field_e:
"barme2".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+ {
+ let input = E { field_union: UnionANoneB::B(B { field_b: 89 }),
field_e: "barme3".to_string() };
+ let output = encode_decode(&input,&schemata[2],&schemata);
+ assert_eq!(input,output);
+ }
+}
