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 2023/06/13 12:00:17 UTC
[avro] branch branch-1.11 updated: AVRO-3772: [Rust] Deserialize Errors for an Unknown Enum Symbol instead of Returning Default (#2280)
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 699598adc AVRO-3772: [Rust] Deserialize Errors for an Unknown Enum Symbol instead of Returning Default (#2280)
699598adc is described below
commit 699598adc93aa00e69e97ed945ee6a914496cd15
Author: Martin Grigorov <ma...@users.noreply.github.com>
AuthorDate: Tue Jun 13 14:59:36 2023 +0300
AVRO-3772: [Rust] Deserialize Errors for an Unknown Enum Symbol instead of Returning Default (#2280)
* AVRO-3772: [Rust] Deserialize Errors for an Unknown Enum Symbol instead of Returning Default
Take into account the field's default value when resolving an enum in a
reader schema
Signed-off-by: Martin Tzvetanov Grigorov <mg...@apache.org>
* AVRO-3772: [Rust] Ignore the length of the enum symbols
Do not take into account the number of the symbols in the reader schema.
There is no such requirement for this in the specification.
Just try to find the written symbol in the list of symbols in the reader
schema.
Signed-off-by: Martin Tzvetanov Grigorov <mg...@apache.org>
---------
Signed-off-by: Martin Tzvetanov Grigorov <mg...@apache.org>
(cherry picked from commit 88985faf13321007d112c27b65dde55c77ca5608)
---
lang/rust/avro/src/lib.rs | 55 ------------
lang/rust/avro/src/schema_compatibility.rs | 132 +++++++++++++++++++++++++++++
lang/rust/avro/src/types.rs | 67 +++++++++------
3 files changed, 172 insertions(+), 82 deletions(-)
diff --git a/lang/rust/avro/src/lib.rs b/lang/rust/avro/src/lib.rs
index af856e906..146a08a2e 100644
--- a/lang/rust/avro/src/lib.rs
+++ b/lang/rust/avro/src/lib.rs
@@ -866,61 +866,6 @@ mod tests {
assert!(reader.next().is_none());
}
- //TODO: move where it fits better
- #[test]
- fn test_enum_resolution() {
- let writer_raw_schema = r#"
- {
- "type": "record",
- "name": "test",
- "fields": [
- {"name": "a", "type": "long", "default": 42},
- {"name": "b", "type": "string"},
- {
- "name": "c",
- "type": {
- "type": "enum",
- "name": "suit",
- "symbols": ["diamonds", "spades", "clubs", "hearts"]
- },
- "default": "spades"
- }
- ]
- }
- "#;
- let reader_raw_schema = r#"
- {
- "type": "record",
- "name": "test",
- "fields": [
- {"name": "a", "type": "long", "default": 42},
- {"name": "b", "type": "string"},
- {
- "name": "c",
- "type": {
- "type": "enum",
- "name": "suit",
- "symbols": ["diamonds", "spades", "ninja", "hearts"]
- },
- "default": "spades"
- }
- ]
- }
- "#;
- let writer_schema = Schema::parse_str(writer_raw_schema).unwrap();
- let reader_schema = Schema::parse_str(reader_raw_schema).unwrap();
- let mut writer = Writer::with_codec(&writer_schema, Vec::new(), Codec::Null);
- let mut record = Record::new(writer.schema()).unwrap();
- record.put("a", 27i64);
- record.put("b", "foo");
- record.put("c", "clubs");
- writer.append(record).unwrap();
- let input = writer.into_inner().unwrap();
- let mut reader = Reader::with_schema(&reader_schema, &input[..]).unwrap();
- assert!(reader.next().unwrap().is_err());
- assert!(reader.next().is_none());
- }
-
//TODO: move where it fits better
#[test]
fn test_enum_no_reader_schema() {
diff --git a/lang/rust/avro/src/schema_compatibility.rs b/lang/rust/avro/src/schema_compatibility.rs
index 2fd50a5bb..b8cd46119 100644
--- a/lang/rust/avro/src/schema_compatibility.rs
+++ b/lang/rust/avro/src/schema_compatibility.rs
@@ -327,6 +327,10 @@ impl SchemaCompatibility {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::{
+ types::{Record, Value},
+ Codec, Reader, Writer,
+ };
use apache_avro_test_helper::TestResult;
fn int_array_schema() -> Schema {
@@ -906,4 +910,132 @@ mod tests {
&read_schema
));
}
+
+ #[test]
+ fn avro_3772_enum_default() -> TestResult {
+ let writer_raw_schema = r#"
+ {
+ "type": "record",
+ "name": "test",
+ "fields": [
+ {"name": "a", "type": "long", "default": 42},
+ {"name": "b", "type": "string"},
+ {
+ "name": "c",
+ "type": {
+ "type": "enum",
+ "name": "suit",
+ "symbols": ["diamonds", "spades", "clubs", "hearts"]
+ },
+ "default": "spades"
+ }
+ ]
+ }
+ "#;
+
+ let reader_raw_schema = r#"
+ {
+ "type": "record",
+ "name": "test",
+ "fields": [
+ {"name": "a", "type": "long", "default": 42},
+ {"name": "b", "type": "string"},
+ {
+ "name": "c",
+ "type": {
+ "type": "enum",
+ "name": "suit",
+ "symbols": ["diamonds", "spades", "ninja", "hearts"]
+ },
+ "default": "spades"
+ }
+ ]
+ }
+ "#;
+ let writer_schema = Schema::parse_str(writer_raw_schema)?;
+ let reader_schema = Schema::parse_str(reader_raw_schema)?;
+ let mut writer = Writer::with_codec(&writer_schema, Vec::new(), Codec::Null);
+ let mut record = Record::new(writer.schema()).unwrap();
+ record.put("a", 27i64);
+ record.put("b", "foo");
+ record.put("c", "clubs");
+ writer.append(record).unwrap();
+ let input = writer.into_inner()?;
+ let mut reader = Reader::with_schema(&reader_schema, &input[..])?;
+ assert_eq!(
+ reader.next().unwrap().unwrap(),
+ Value::Record(vec![
+ ("a".to_string(), Value::Long(27)),
+ ("b".to_string(), Value::String("foo".to_string())),
+ ("c".to_string(), Value::Enum(1, "spades".to_string())),
+ ])
+ );
+ assert!(reader.next().is_none());
+
+ Ok(())
+ }
+
+ #[test]
+ fn avro_3772_enum_default_less_symbols() -> TestResult {
+ let writer_raw_schema = r#"
+ {
+ "type": "record",
+ "name": "test",
+ "fields": [
+ {"name": "a", "type": "long", "default": 42},
+ {"name": "b", "type": "string"},
+ {
+ "name": "c",
+ "type": {
+ "type": "enum",
+ "name": "suit",
+ "symbols": ["diamonds", "spades", "clubs", "hearts"]
+ },
+ "default": "spades"
+ }
+ ]
+ }
+ "#;
+
+ let reader_raw_schema = r#"
+ {
+ "type": "record",
+ "name": "test",
+ "fields": [
+ {"name": "a", "type": "long", "default": 42},
+ {"name": "b", "type": "string"},
+ {
+ "name": "c",
+ "type": {
+ "type": "enum",
+ "name": "suit",
+ "symbols": ["hearts", "spades"]
+ },
+ "default": "spades"
+ }
+ ]
+ }
+ "#;
+ let writer_schema = Schema::parse_str(writer_raw_schema)?;
+ let reader_schema = Schema::parse_str(reader_raw_schema)?;
+ let mut writer = Writer::with_codec(&writer_schema, Vec::new(), Codec::Null);
+ let mut record = Record::new(writer.schema()).unwrap();
+ record.put("a", 27i64);
+ record.put("b", "foo");
+ record.put("c", "hearts");
+ writer.append(record).unwrap();
+ let input = writer.into_inner()?;
+ let mut reader = Reader::with_schema(&reader_schema, &input[..])?;
+ assert_eq!(
+ reader.next().unwrap().unwrap(),
+ Value::Record(vec![
+ ("a".to_string(), Value::Long(27)),
+ ("b".to_string(), Value::String("foo".to_string())),
+ ("c".to_string(), Value::Enum(0, "hearts".to_string())),
+ ])
+ );
+ assert!(reader.next().is_none());
+
+ Ok(())
+ }
}
diff --git a/lang/rust/avro/src/types.rs b/lang/rust/avro/src/types.rs
index 0adf72112..6ecadda0a 100644
--- a/lang/rust/avro/src/types.rs
+++ b/lang/rust/avro/src/types.rs
@@ -566,7 +566,7 @@ impl Value {
pub fn resolve(self, schema: &Schema) -> AvroResult<Self> {
let enclosing_namespace = schema.namespace();
let rs = ResolvedSchema::try_from(schema)?;
- self.resolve_internal(schema, rs.get_names(), &enclosing_namespace)
+ self.resolve_internal(schema, rs.get_names(), &enclosing_namespace, &None)
}
/// Attempt to perform schema resolution on the value, with the given
@@ -578,7 +578,7 @@ impl Value {
pub fn resolve_schemata(self, schema: &Schema, schemata: Vec<&Schema>) -> AvroResult<Self> {
let enclosing_namespace = schema.namespace();
let rs = ResolvedSchema::try_from(schemata)?;
- self.resolve_internal(schema, rs.get_names(), &enclosing_namespace)
+ self.resolve_internal(schema, rs.get_names(), &enclosing_namespace, &None)
}
fn resolve_internal(
@@ -586,6 +586,7 @@ impl Value {
schema: &Schema,
names: &NamesRef,
enclosing_namespace: &Namespace,
+ field_default: &Option<JsonValue>,
) -> AvroResult<Self> {
// Check if this schema is a union, and if the reader schema is not.
if SchemaKind::from(&self) == SchemaKind::Union
@@ -605,7 +606,7 @@ impl Value {
if let Some(resolved) = names.get(&name) {
debug!("Resolved {:?}", name);
- self.resolve_internal(resolved, names, &name.namespace)
+ self.resolve_internal(resolved, names, &name.namespace, field_default)
} else {
error!("Failed to resolve schema {:?}", name);
Err(Error::SchemaResolutionError(name.clone()))
@@ -620,8 +621,12 @@ impl Value {
Schema::Bytes => self.resolve_bytes(),
Schema::String => self.resolve_string(),
Schema::Fixed(FixedSchema { size, .. }) => self.resolve_fixed(size),
- Schema::Union(ref inner) => self.resolve_union(inner, names, enclosing_namespace),
- Schema::Enum(EnumSchema { ref symbols, .. }) => self.resolve_enum(symbols),
+ Schema::Union(ref inner) => {
+ self.resolve_union(inner, names, enclosing_namespace, field_default)
+ }
+ Schema::Enum(EnumSchema { ref symbols, .. }) => {
+ self.resolve_enum(symbols, field_default)
+ }
Schema::Array(ref inner) => self.resolve_array(inner, names, enclosing_namespace),
Schema::Map(ref inner) => self.resolve_map(inner, names, enclosing_namespace),
Schema::Record(RecordSchema { ref fields, .. }) => {
@@ -840,31 +845,36 @@ impl Value {
}
}
- fn resolve_enum(self, symbols: &[String]) -> Result<Self, Error> {
+ fn resolve_enum(
+ self,
+ symbols: &[String],
+ field_default: &Option<JsonValue>,
+ ) -> Result<Self, Error> {
let validate_symbol = |symbol: String, symbols: &[String]| {
if let Some(index) = symbols.iter().position(|item| item == &symbol) {
Ok(Value::Enum(index as u32, symbol))
} else {
- Err(Error::GetEnumDefault {
- symbol,
- symbols: symbols.into(),
- })
+ match field_default {
+ Some(JsonValue::String(default)) => {
+ if let Some(index) = symbols.iter().position(|item| item == default) {
+ Ok(Value::Enum(index as u32, default.clone()))
+ } else {
+ Err(Error::GetEnumDefault {
+ symbol,
+ symbols: symbols.into(),
+ })
+ }
+ }
+ _ => Err(Error::GetEnumDefault {
+ symbol,
+ symbols: symbols.into(),
+ }),
+ }
}
};
match self {
- Value::Enum(raw_index, s) => {
- let index = usize::try_from(raw_index)
- .map_err(|e| Error::ConvertU32ToUsize(e, raw_index))?;
- if (0..=symbols.len()).contains(&index) {
- validate_symbol(s, symbols)
- } else {
- Err(Error::GetEnumValue {
- index,
- nsymbols: symbols.len(),
- })
- }
- }
+ Value::Enum(_raw_index, s) => validate_symbol(s, symbols),
Value::String(s) => validate_symbol(s, symbols),
other => Err(Error::GetEnum(other.into())),
}
@@ -875,6 +885,7 @@ impl Value {
schema: &UnionSchema,
names: &NamesRef,
enclosing_namespace: &Namespace,
+ field_default: &Option<JsonValue>,
) -> Result<Self, Error> {
let v = match self {
// Both are unions case.
@@ -889,7 +900,7 @@ impl Value {
Ok(Value::Union(
i as u32,
- Box::new(v.resolve_internal(inner, names, enclosing_namespace)?),
+ Box::new(v.resolve_internal(inner, names, enclosing_namespace, field_default)?),
))
}
@@ -903,7 +914,7 @@ impl Value {
Value::Array(items) => Ok(Value::Array(
items
.into_iter()
- .map(|item| item.resolve_internal(schema, names, enclosing_namespace))
+ .map(|item| item.resolve_internal(schema, names, enclosing_namespace, &None))
.collect::<Result<_, _>>()?,
)),
other => Err(Error::GetArray {
@@ -925,7 +936,7 @@ impl Value {
.into_iter()
.map(|(key, value)| {
value
- .resolve_internal(schema, names, enclosing_namespace)
+ .resolve_internal(schema, names, enclosing_namespace, &None)
.map(|value| (key, value))
})
.collect::<Result<_, _>>()?,
@@ -963,7 +974,8 @@ impl Value {
None => match field.default {
Some(ref value) => match field.schema {
Schema::Enum(EnumSchema { ref symbols, .. }) => {
- Value::from(value.clone()).resolve_enum(symbols)?
+ Value::from(value.clone())
+ .resolve_enum(symbols, &field.default.clone())?
}
Schema::Union(ref union_schema) => {
let first = &union_schema.variants()[0];
@@ -977,6 +989,7 @@ impl Value {
first,
names,
enclosing_namespace,
+ &field.default,
)?),
),
}
@@ -989,7 +1002,7 @@ impl Value {
},
};
value
- .resolve_internal(&field.schema, names, enclosing_namespace)
+ .resolve_internal(&field.schema, names, enclosing_namespace, &field.default)
.map(|value| (field.name.clone(), value))
})
.collect::<Result<Vec<_>, _>>()?;