This is an automated email from the ASF dual-hosted git repository. Kriskras99 pushed a commit to branch feat/improve_schema_parser in repository https://gitbox.apache.org/repos/asf/avro-rs.git
commit fbd9c72799b5c566ffc5dad0bf28a50040f2622b Author: Kriskras99 <[email protected]> AuthorDate: Fri Sep 4 21:10:00 2026 +0200 feat: Make schema parser reuse the JSON allocations Also includes a fix for https://github.com/apache/avro-rs/issues/654 --- avro/src/error.rs | 27 +++++ avro/src/reader/block.rs | 4 +- avro/src/schema/mod.rs | 97 +++++++++++++---- avro/src/schema/name.rs | 21 ++-- avro/src/schema/parser.rs | 226 ++++++++++++++++++---------------------- avro/src/schema/record/field.rs | 44 +++----- avro/src/types.rs | 14 +-- avro/src/util.rs | 75 +++++++++---- 8 files changed, 293 insertions(+), 215 deletions(-) diff --git a/avro/src/error.rs b/avro/src/error.rs index 6723340..311a647 100644 --- a/avro/src/error.rs +++ b/avro/src/error.rs @@ -304,6 +304,9 @@ pub enum Details { #[error("No `name` field")] GetNameField, + #[error("Expected a string for the `name` field, got a {0}")] + GetNamespaceFieldWrongType(&'static str), + #[error("No `name` in record field")] GetNameFieldFromRecord, @@ -431,9 +434,21 @@ pub enum Details { #[error("No `fields` in record")] GetRecordFieldsJson, + #[error("Expected an object in the array of the `fields` field, got a {0}")] + GetRecordFieldsArrayInvalidType(&'static str), + + #[error("Expected an array of objects for the `fields` field, got a {0}")] + GetRecordFieldsInvalidType(&'static str), + #[error("No `symbols` field in enum")] GetEnumSymbolsField, + #[error("Expected an array of strings for the `symbols` field, got a {0}")] + GetEnumSymbolsFieldInvalidType(&'static str), + + #[error("Expected a string in the array of the `symbols` field, got a {0}")] + GetEnumSymbolsFieldArrayInvalidType(&'static str), + #[error("Unable to parse `symbols` in enum")] GetEnumSymbols, @@ -487,6 +502,18 @@ pub enum Details { #[error("Fixed schema has no `size`")] GetFixedSizeField, + #[error("Expected an unsigned integer for the `size` field, got a {0}")] + GetFixedSizeFieldInvalidType(&'static str), + + #[error("Expected an array of strings for the `aliases` field, got a {0}")] + GetAliasesFieldInvalidType(&'static str), + + #[error("Expected a string in the array for the `aliases` field, got a {0}")] + GetAliasesFieldArrayInvalidType(&'static str), + + #[error("Expected a string for the `{0}` field, got a {1}")] + GetStringInvalidType(&'static str, &'static str), + #[deprecated(since = "0.22.0", note = "This error variant is not generated anymore")] #[error("Fixed schema's default value length ({0}) does not match its size ({1})")] FixedDefaultLenSizeMismatch(usize, u64), diff --git a/avro/src/reader/block.rs b/avro/src/reader/block.rs index b73e01d..73d045a 100644 --- a/avro/src/reader/block.rs +++ b/avro/src/reader/block.rs @@ -288,9 +288,9 @@ impl<'r, R: Read> Block<'r, R> { &HashMap::new(), )?; self.names_refs = names.into_iter().map(|(n, s)| (n, s.clone())).collect(); - self.writer_schema = Schema::parse_with_names(&json, self.names_refs.clone())?; + self.writer_schema = Schema::parse_with_names(json, self.names_refs.clone())?; } else { - self.writer_schema = Schema::parse(&json)?; + self.writer_schema = Schema::parse(json)?; let mut names = HashMap::new(); resolve_names(&self.writer_schema, &mut names, None, &HashMap::new())?; self.names_refs = names.into_iter().map(|(n, s)| (n, s.clone())).collect(); diff --git a/avro/src/schema/mod.rs b/avro/src/schema/mod.rs index fe3d72e..fe58f07 100644 --- a/avro/src/schema/mod.rs +++ b/avro/src/schema/mod.rs @@ -546,7 +546,16 @@ impl Schema { let json = json.as_ref(); let schema: JsonValue = serde_json::from_str(json).map_err(Details::ParseSchemaJson)?; if let JsonValue::Object(inner) = &schema { - let name = Name::parse(inner, None)?; + // Only clone the keys needed for the name parsing, can be a significant time/memory + // save on large schemas + let mut name_json = Map::with_capacity(2); + if let Some(v) = inner.get("name") { + name_json.insert("name".into(), v.clone()); + } + if let Some(v) = inner.get("namespace") { + name_json.insert("namespace".into(), v.clone()); + } + let name = Name::parse(&mut name_json, None)?; let previous_value = input_schemas.insert(name.clone(), schema); if previous_value.is_some() { return Err(Details::NameCollision(name.fullname(None)).into()); @@ -588,7 +597,16 @@ impl Schema { let json = json.as_ref(); let schema: JsonValue = serde_json::from_str(json).map_err(Details::ParseSchemaJson)?; if let JsonValue::Object(inner) = &schema { - let name = Name::parse(inner, None)?; + // Only clone the keys needed for the name parsing, can be a significant time/memory + // save on large schemas + let mut name_json = Map::with_capacity(2); + if let Some(v) = inner.get("name") { + name_json.insert("name".into(), v.clone()); + } + if let Some(v) = inner.get("namespace") { + name_json.insert("namespace".into(), v.clone()); + } + let name = Name::parse(&mut name_json, None)?; if let Some(_previous) = input_schemas.insert(name.clone(), schema) { return Err(Details::NameCollision(name.fullname(None)).into()); } @@ -605,7 +623,7 @@ impl Schema { parser.parse_input_schemas()?; let value = serde_json::from_str(schema).map_err(Details::ParseSchemaJson)?; - let schema = parser.parse(&value, None)?; + let schema = parser.parse(value, None)?; let schemata = parser.parse_list()?; Ok((schema, schemata)) } @@ -620,14 +638,14 @@ impl Schema { } /// Parses an Avro schema from JSON. - pub fn parse(value: &JsonValue) -> AvroResult<Schema> { + pub fn parse(value: JsonValue) -> AvroResult<Schema> { let mut parser = Parser::default(); parser.parse(value, None) } /// Parses an Avro schema from JSON. /// Any `Schema::Ref`s must be known in the `names` map. - pub(crate) fn parse_with_names(value: &JsonValue, names: Names) -> AvroResult<Schema> { + pub(crate) fn parse_with_names(value: JsonValue, names: Names) -> AvroResult<Schema> { let mut parser = Parser::new(HashMap::with_capacity(1), Vec::with_capacity(1), names); parser.parse(value, None) } @@ -3094,7 +3112,7 @@ mod tests { ] }); - let parse_result = Schema::parse(&schema); + let parse_result = Schema::parse(schema); assert!( parse_result.is_ok(), "parse result must be ok, got: {parse_result:?}" @@ -3341,7 +3359,7 @@ mod tests { } ); - let parse_result = Schema::parse(&schema); + let parse_result = Schema::parse(schema); assert!( parse_result.is_ok(), "parse result must be ok, got: {parse_result:?}" @@ -3631,7 +3649,7 @@ mod tests { }; // Serialize using the writer schema. - let writer_schema = Schema::parse(&writer_schema)?; + let writer_schema = Schema::parse(writer_schema)?; let avro_value = crate::to_value(s)?; assert!( avro_value.validate(&writer_schema), @@ -3642,7 +3660,7 @@ mod tests { .write_value_to_vec(avro_value)?; // Now, attempt to deserialize using the reader schema. - let reader_schema = Schema::parse(&reader_schema)?; + let reader_schema = Schema::parse(reader_schema)?; let mut x = &datum[..]; // Deserialization should succeed and we should be able to resolve the schema. @@ -4328,7 +4346,7 @@ mod tests { "precision": 9, "scale": 2 }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert!(matches!( parse_result, Schema::Decimal(DecimalSchema { @@ -4345,7 +4363,7 @@ mod tests { "name": "LongDecimal", "logicalType": "decimal" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; // assert!(matches!(parse_result, Schema::Long)); assert_eq!(parse_result, Schema::Long); @@ -4361,7 +4379,7 @@ mod tests { "name": "StringUUID", "logicalType": "uuid" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert_eq!(parse_result, Schema::Uuid(UuidSchema::String)); Ok(()) @@ -4376,7 +4394,7 @@ mod tests { "size": 16, "logicalType": "uuid" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert_eq!( parse_result, Schema::Uuid(UuidSchema::Fixed(FixedSchema { @@ -4402,7 +4420,7 @@ mod tests { "name": "BytesUUID", "logicalType": "uuid" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert_eq!(parse_result, Schema::Uuid(UuidSchema::Bytes)); Ok(()) @@ -4417,7 +4435,7 @@ mod tests { "size": 6, "logicalType": "uuid" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert_eq!( parse_result, @@ -4445,7 +4463,7 @@ mod tests { "name": "LongTimestampMillis", "logicalType": "timestamp-millis" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert_eq!(parse_result, Schema::TimestampMillis); // int timestamp-millis, represents as native complex type. @@ -4455,7 +4473,7 @@ mod tests { "name": "IntTimestampMillis", "logicalType": "timestamp-millis" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert_eq!(parse_result, Schema::Int); Ok(()) @@ -4470,7 +4488,7 @@ mod tests { "name": "BytesLog", "logicalType": "custom" }); - let parse_result = Schema::parse(&schema)?; + let parse_result = Schema::parse(schema)?; assert_eq!(parse_result, Schema::Bytes); assert_eq!(parse_result.custom_attributes(), None); @@ -5234,4 +5252,47 @@ mod tests { Ok(()) } + + #[test] + fn avro_rs_654_unknown_logical_type_must_survive_roundtrip() -> TestResult { + let schema_str = r#"{ + "type": "array", + "logicalType": "map", + "items": { + "type": "record", + "name": "k12_v13", + "fields": [ + { + "name": "key", + "type": "int", + "field-id": 12 + }, + { + "name": "value", + "type": "string", + "field-id": 13 + } + ] + } + }"#; + let Schema::Array(schema) = Schema::parse_str(schema_str)? else { + panic!("This must be an array schema") + }; + assert_eq!( + schema.attributes.get("logicalType").unwrap(), + &serde_json::Value::String("map".into()) + ); + + let schema_str_2 = serde_json::to_string(&Schema::Array(schema))?; + + let Schema::Array(schema) = Schema::parse_str(&schema_str_2)? else { + panic!("This must be an array schema") + }; + assert_eq!( + schema.attributes.get("logicalType").unwrap(), + &serde_json::Value::String("map".into()) + ); + + Ok(()) + } } diff --git a/avro/src/schema/name.rs b/avro/src/schema/name.rs index bfb69f2..d290863 100644 --- a/avro/src/schema/name.rs +++ b/avro/src/schema/name.rs @@ -22,6 +22,7 @@ use std::collections::HashMap; use std::fmt::{Debug, Display, Formatter}; use std::str::FromStr; +use crate::util::JsonValueDescriber; use crate::{ AvroResult, Error, Schema, error::Details, @@ -110,14 +111,18 @@ impl Name { /// Parse a `serde_json::Value` into a `Name`. pub(crate) fn parse( - complex: &Map<String, Value>, + complex: &mut Map<String, Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Self> { - let name_field = complex.name().ok_or(Details::GetNameField)?; - Self::new_with_enclosing_namespace( - name_field, - complex.string("namespace").or(enclosing_namespace), - ) + let name_field = complex.name()?; + let namespace = match complex.remove("namespace") { + Some(Value::String(s)) => Some(s), + Some(Value::Null) | None => None, + Some(value) => { + return Err(Details::GetNamespaceFieldWrongType(value.description()).into()); + } + }; + Self::new_with_enclosing_namespace(name_field, namespace.as_deref().or(enclosing_namespace)) } pub fn name(&self) -> &str { @@ -253,8 +258,8 @@ impl<'de> Deserialize<'de> for Name { { Value::deserialize(deserializer).and_then(|value| { use serde::de::Error; - if let Value::Object(json) = value { - Name::parse(&json, None).map_err(Error::custom) + if let Value::Object(mut json) = value { + Name::parse(&mut json, None).map_err(Error::custom) } else { Err(Error::custom(format!("Expected a JSON object: {value:?}"))) } diff --git a/avro/src/schema/parser.rs b/avro/src/schema/parser.rs index 7f74af6..8b8f877 100644 --- a/avro/src/schema/parser.rs +++ b/avro/src/schema/parser.rs @@ -21,8 +21,7 @@ use crate::schema::{ MapSchema, Name, Names, NamespaceRef, Precision, RecordField, RecordSchema, Scale, Schema, SchemaKind, UnionSchema, UuidSchema, }; -use crate::types; -use crate::util::MapHelper; +use crate::util::{JsonValueDescriber, MapHelper}; use crate::validator::validate_enum_symbol_name; use crate::{AvroResult, Error}; use log::{debug, error, warn}; @@ -61,7 +60,7 @@ impl Parser { /// Create a `Schema` from a string representing a JSON Avro schema. pub(super) fn parse_str(&mut self, input: &str) -> AvroResult<Schema> { let value = serde_json::from_str(input).map_err(Details::ParseSchemaJson)?; - self.parse(&value, None) + self.parse(value, None) } /// Create an array of `Schema`s from an iterator of JSON Avro schemas. @@ -94,9 +93,9 @@ impl Parser { .input_schemas .remove_entry(&next_name) .expect("Key unexpectedly missing"); - let parsed = self.parse(&value, None)?; - self.parsed_schemas - .insert(self.get_schema_type_name(name, &value)?, parsed); + let full_name = self.get_schema_type_name(name, &value)?; + let parsed = self.parse(value, None)?; + self.parsed_schemas.insert(full_name, parsed); } Ok(()) } @@ -104,13 +103,13 @@ impl Parser { /// Create a `Schema` from a `serde_json::Value` representing a JSON Avro schema. pub(super) fn parse( &mut self, - value: &Value, + value: Value, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { - match *value { - Value::String(ref t) => self.parse_known_schema(t.as_str(), enclosing_namespace), - Value::Object(ref data) => self.parse_complex(data, enclosing_namespace), - Value::Array(ref data) => self.parse_union(data, enclosing_namespace), + match value { + Value::String(t) => self.parse_known_schema(t.as_str(), enclosing_namespace), + Value::Object(data) => self.parse_complex(data, enclosing_namespace), + Value::Array(data) => self.parse_union(data, enclosing_namespace), _ => Err(Details::ParseSchemaFromValidJson.into()), } } @@ -149,17 +148,6 @@ impl Parser { name: &str, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { - fn get_schema_ref(parsed: &Schema) -> Schema { - match parsed { - &Schema::Record(RecordSchema { ref name, .. }) - | &Schema::Enum(EnumSchema { ref name, .. }) - | &Schema::Fixed(FixedSchema { ref name, .. }) => { - Schema::Ref { name: name.clone() } - } - _ => parsed.clone(), - } - } - let fully_qualified_name = Name::new_with_enclosing_namespace(name, enclosing_namespace)?; if self.parsed_schemas.contains_key(&fully_qualified_name) { @@ -195,13 +183,11 @@ impl Parser { })?; // parsing a full schema from inside another schema. Other full schema will not inherit namespace - let parsed = self.parse(&value, None)?; - self.parsed_schemas.insert( - self.get_schema_type_name(fully_qualified_name, &value)?, - parsed.clone(), - ); + let full_name = self.get_schema_type_name(fully_qualified_name, &value)?; + let parsed = self.parse(value, None)?; + self.parsed_schemas.insert(full_name.clone(), parsed); - Ok(get_schema_ref(&parsed)) + Ok(Schema::Ref { name: full_name }) } fn get_decimal_integer( @@ -250,16 +236,16 @@ impl Parser { /// e.g: `{"type": {"type": "string"}}` pub(super) fn parse_complex( &mut self, - complex: &Map<String, Value>, + mut complex: Map<String, Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { // Try to parse this as a native complex type. fn parse_as_native_complex( - complex: &Map<String, Value>, + mut complex: Map<String, Value>, parser: &mut Parser, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { - match complex.get("type") { + match complex.remove("type") { Some(value) => match value { Value::String(s) if s == "fixed" => { parser.parse_fixed(complex, enclosing_namespace) @@ -296,15 +282,17 @@ impl Parser { } } - match complex.get("logicalType") { - Some(Value::String(t)) => match t.as_str() { + match complex.remove_entry("logicalType") { + Some((key, Value::String(t))) => match t.as_str() { "decimal" => { return try_convert_to_logical_type( "decimal", - parse_as_native_complex(complex, self, enclosing_namespace)?, + // TODO: See if we can avoid this clone, although if this really is a decimal + // the clone is cheap enough not to be a problem + parse_as_native_complex(complex.clone(), self, enclosing_namespace)?, &[SchemaKind::Fixed, SchemaKind::Bytes], |inner| -> AvroResult<Schema> { - match self.parse_precision_and_scale(complex) { + match self.parse_precision_and_scale(&complex) { Ok((precision, scale)) => Ok(Schema::Decimal(DecimalSchema { precision, scale, @@ -450,15 +438,18 @@ impl Parser { } // In this case, of an unknown logical type, we just pass through the underlying // type. - _ => {} + _ => { + // re-insert unknown logical type + complex.insert(key, Value::String(t)); + } }, // The spec says to ignore invalid logical types and just pass through the // underlying type. It is unclear whether that applies to this case or not, where the // `logicalType` is not a string. - Some(value) => return Err(Details::GetLogicalTypeFieldType(value.clone()).into()), + Some((_, value)) => return Err(Details::GetLogicalTypeFieldType(value.clone()).into()), _ => {} } - match complex.get("type") { + match complex.remove("type") { Some(Value::String(t)) => match t.as_str() { "record" => self.parse_record(complex, enclosing_namespace), "enum" => self.parse_enum(complex, enclosing_namespace), @@ -539,20 +530,20 @@ impl Parser { /// Parse a `serde_json::Value` representing an Avro record type into a `Schema`. fn parse_record( &mut self, - complex: &Map<String, Value>, + mut complex: Map<String, Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { - let fields_opt = complex.get("fields"); + let fields_opt = complex.remove("fields"); if fields_opt.is_none() - && let Some(seen) = self.get_already_seen_schema(complex, enclosing_namespace) + && let Some(seen) = self.get_already_seen_schema(&complex, enclosing_namespace) { return Ok(seen.clone()); } - let fully_qualified_name = Name::parse(complex, enclosing_namespace)?; + let fully_qualified_name = Name::parse(&mut complex, enclosing_namespace)?; let aliases = - self.fix_aliases_namespace(complex.aliases(), fully_qualified_name.namespace())?; + self.fix_aliases_namespace(complex.aliases()?, fully_qualified_name.namespace())?; let mut lookup = BTreeMap::new(); @@ -560,16 +551,19 @@ impl Parser { debug!("Going to parse record schema: {fully_qualified_name:?}"); - let fields: Vec<RecordField> = fields_opt - .and_then(|fields| fields.as_array()) - .ok_or_else(|| Error::new(Details::GetRecordFieldsJson)) - .and_then(|fields| { - fields - .iter() - .filter_map(|field| field.as_object()) - .map(|field| RecordField::parse(field, self, &fully_qualified_name)) - .collect::<Result<_, _>>() - })?; + let fields = match fields_opt { + Some(Value::Array(array)) => array + .into_iter() + .map(|v| match v { + Value::Object(o) => RecordField::parse(o, self, &fully_qualified_name), + _ => Err(Details::GetRecordFieldsArrayInvalidType(v.description()).into()), + }) + .collect::<Result<Vec<_>, _>>()?, + Some(value) => { + return Err(Details::GetRecordFieldsInvalidType(value.description()).into()); + } + None => return Err(Details::GetRecordFieldsJson.into()), + }; for (position, field) in fields.iter().enumerate() { if let Some(_old) = lookup.insert(field.name.clone(), position) { @@ -584,60 +578,51 @@ impl Parser { let schema = Schema::Record(RecordSchema { name: fully_qualified_name.clone(), aliases: aliases.clone(), - doc: complex.doc(), + doc: complex.doc()?, fields, lookup, - attributes: self.get_custom_attributes(complex, &["fields"]), + attributes: self.get_custom_attributes(complex), }); self.register_parsed_schema(&fully_qualified_name, &schema, &aliases); Ok(schema) } - fn get_custom_attributes( - &self, - complex: &Map<String, Value>, - excluded: &[&'static str], - ) -> BTreeMap<String, Value> { - let mut custom_attributes: BTreeMap<String, Value> = BTreeMap::new(); - for (key, value) in complex { - match key.as_str() { - "type" | "name" | "namespace" | "doc" | "aliases" | "logicalType" => continue, - candidate if excluded.contains(&candidate) => continue, - _ => custom_attributes.insert(key.clone(), value.clone()), - }; - } - custom_attributes + fn get_custom_attributes(&self, complex: Map<String, Value>) -> BTreeMap<String, Value> { + complex.into_iter().collect() } /// Parse a `serde_json::Value` representing a Avro enum type into a `Schema`. fn parse_enum( &mut self, - complex: &Map<String, Value>, + mut complex: Map<String, Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { - let symbols_opt = complex.get("symbols"); + let symbols_opt = complex.remove("symbols"); if symbols_opt.is_none() - && let Some(seen) = self.get_already_seen_schema(complex, enclosing_namespace) + && let Some(seen) = self.get_already_seen_schema(&complex, enclosing_namespace) { return Ok(seen.clone()); } - let fully_qualified_name = Name::parse(complex, enclosing_namespace)?; + let fully_qualified_name = Name::parse(&mut complex, enclosing_namespace)?; let aliases = - self.fix_aliases_namespace(complex.aliases(), fully_qualified_name.namespace())?; - - let symbols: Vec<String> = symbols_opt - .and_then(|v| v.as_array()) - .ok_or_else(|| Error::from(Details::GetEnumSymbolsField)) - .and_then(|symbols| { - symbols - .iter() - .map(|symbol| symbol.as_str().map(|s| s.to_string())) - .collect::<Option<_>>() - .ok_or_else(|| Error::from(Details::GetEnumSymbols)) - })?; + self.fix_aliases_namespace(complex.aliases()?, fully_qualified_name.namespace())?; + + let symbols = match symbols_opt { + Some(Value::Array(array)) => array + .into_iter() + .map(|v| match v { + Value::String(s) => Ok(s), + _ => Err(Error::new(Details::GetEnumSymbolsFieldArrayInvalidType( + v.description(), + ))), + }) + .collect::<Result<Vec<_>, _>>(), + Some(value) => Err(Details::GetEnumSymbolsFieldInvalidType(value.description()).into()), + None => Err(Details::GetEnumSymbolsField.into()), + }?; let mut existing_symbols: HashSet<&String> = HashSet::with_capacity(symbols.len()); for symbol in symbols.iter() { @@ -651,35 +636,24 @@ impl Parser { existing_symbols.insert(symbol); } - let mut default: Option<String> = None; - if let Some(value) = complex.get("default") { - if let Value::String(ref s) = *value { - default = Some(s.clone()); - } else { - return Err(Details::EnumDefaultWrongType(value.clone()).into()); - } - } - - if let Some(ref value) = default { - let resolved = types::Value::from(value.clone()) - .resolve_enum(&symbols, &Some(value.to_string()), None) - .is_ok(); - if !resolved { - return Err(Details::GetEnumDefault { - symbol: value.to_string(), - symbols, + let default = match complex.remove("default") { + Some(Value::String(s)) => { + if !symbols.contains(&s) { + return Err(Details::GetEnumDefault { symbol: s, symbols }.into()); } - .into()); + Some(s) } - } + Some(v) => return Err(Details::EnumDefaultWrongType(v).into()), + None => None, + }; let schema = Schema::Enum(EnumSchema { name: fully_qualified_name.clone(), aliases: aliases.clone(), - doc: complex.doc(), + doc: complex.doc()?, symbols, default, - attributes: self.get_custom_attributes(complex, &["symbols", "default"]), + attributes: self.get_custom_attributes(complex), }); self.register_parsed_schema(&fully_qualified_name, &schema, &aliases); @@ -690,44 +664,44 @@ impl Parser { /// Parse a `serde_json::Value` representing a Avro array type into a `Schema`. fn parse_array( &mut self, - complex: &Map<String, Value>, + mut complex: Map<String, Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { let items = complex - .get("items") + .remove("items") .ok_or_else(|| Details::GetArrayItemsField.into()) .and_then(|items| self.parse(items, enclosing_namespace))?; Ok(Schema::Array(ArraySchema { items: Box::new(items), - attributes: self.get_custom_attributes(complex, &["items"]), + attributes: self.get_custom_attributes(complex), })) } /// Parse a `serde_json::Value` representing a Avro map type into a `Schema`. fn parse_map( &mut self, - complex: &Map<String, Value>, + mut complex: Map<String, Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { let types = complex - .get("values") + .remove("values") .ok_or_else(|| Details::GetMapValuesField.into()) .and_then(|types| self.parse(types, enclosing_namespace))?; Ok(Schema::Map(MapSchema { types: Box::new(types), - attributes: self.get_custom_attributes(complex, &["values"]), + attributes: self.get_custom_attributes(complex), })) } /// Parse a `serde_json::Value` representing a Avro union type into a `Schema`. fn parse_union( &mut self, - items: &[Value], + items: Vec<Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { items - .iter() + .into_iter() .map(|v| self.parse(v, enclosing_namespace)) .collect::<Result<Vec<_>, _>>() .and_then(|schemas| { @@ -751,38 +725,36 @@ impl Parser { /// Parse a `serde_json::Value` representing a Avro fixed type into a `Schema`. fn parse_fixed( &mut self, - complex: &Map<String, Value>, + mut complex: Map<String, Value>, enclosing_namespace: NamespaceRef, ) -> AvroResult<Schema> { - let size_opt = complex.get("size"); + let size_opt = complex.remove("size"); if size_opt.is_none() - && let Some(seen) = self.get_already_seen_schema(complex, enclosing_namespace) + && let Some(seen) = self.get_already_seen_schema(&complex, enclosing_namespace) { return Ok(seen.clone()); } - let doc = complex.get("doc").and_then(|v| match &v { - &Value::String(docstr) => Some(docstr.clone()), - _ => None, - }); + let doc = complex.string("doc")?; let size = match size_opt { - Some(size) => size + Some(Value::Number(size)) => size .as_u64() - .ok_or_else(|| Details::GetFixedSizeFieldPositive(size.clone())), + .ok_or_else(|| Details::GetFixedSizeFieldPositive(Value::Number(size.clone()))), + Some(v) => Err(Details::GetFixedSizeFieldInvalidType(v.description())), None => Err(Details::GetFixedSizeField), }?; let size = usize::try_from(size).map_err(|e| Details::ConvertU64ToUsize(e, size))?; - let fully_qualified_name = Name::parse(complex, enclosing_namespace)?; + let fully_qualified_name = Name::parse(&mut complex, enclosing_namespace)?; let aliases = - self.fix_aliases_namespace(complex.aliases(), fully_qualified_name.namespace())?; + self.fix_aliases_namespace(complex.aliases()?, fully_qualified_name.namespace())?; let schema = Schema::Fixed(FixedSchema { name: fully_qualified_name.clone(), aliases: aliases.clone(), doc, size, - attributes: self.get_custom_attributes(complex, &["size"]), + attributes: self.get_custom_attributes(complex), }); self.register_parsed_schema(&fully_qualified_name, &schema, &aliases); @@ -817,7 +789,7 @@ impl Parser { fn get_schema_type_name(&self, name: Name, value: &Value) -> AvroResult<Name> { match value.get("type") { - Some(Value::Object(complex_type)) => match complex_type.name() { + Some(Value::Object(complex_type)) => match complex_type.name_ref()? { // Propagate the validation error if the nested `type` name is // not a valid Avro name, rather than panicking on `unwrap()`. Some(type_name) => Name::new(type_name), diff --git a/avro/src/schema/record/field.rs b/avro/src/schema/record/field.rs index 2bd6468..38e0880 100644 --- a/avro/src/schema/record/field.rs +++ b/avro/src/schema/record/field.rs @@ -83,15 +83,17 @@ impl Debug for RecordField { impl RecordField { /// Parse a `serde_json::Value` into a `RecordField`. pub(crate) fn parse( - field: &Map<String, Value>, + mut field: Map<String, Value>, parser: &mut Parser, enclosing_record: &Name, ) -> AvroResult<Self> { - let name = field.name().ok_or(Details::GetNameFieldFromRecord)?; + let name = field.name()?; - validate_record_field_name(name)?; + validate_record_field_name(&name)?; - let ty = field.get("type").ok_or(Details::GetRecordFieldTypeField)?; + let ty = field + .remove("type") + .ok_or(Details::GetRecordFieldTypeField)?; let schema = parser.parse(ty, enclosing_record.namespace())?; if let Some(logical_type) = field.get("logicalType") { @@ -100,34 +102,23 @@ impl RecordField { ); } - let default = field.get("default").cloned(); + let default = field.remove("default"); Self::resolve_default_value( &schema, - name, + &name, &enclosing_record.fullname(None), parser.get_parsed_schemas(), &default, )?; - let aliases = field - .get("aliases") - .and_then(|aliases| { - aliases.as_array().map(|aliases| { - aliases - .iter() - .flat_map(|alias| alias.as_str()) - .map(|alias| alias.to_string()) - .collect::<Vec<String>>() - }) - }) - .unwrap_or_default(); + let aliases = field.aliases()?.unwrap_or_default(); Ok(RecordField { - name: name.into(), - doc: field.doc(), + name, + doc: field.doc()?, default, aliases, - custom_attributes: RecordField::get_field_custom_attributes(field), + custom_attributes: field.into_iter().collect(), schema, }) } @@ -184,17 +175,6 @@ impl RecordField { Ok(()) } - fn get_field_custom_attributes(field: &Map<String, Value>) -> BTreeMap<String, Value> { - let mut custom_attributes: BTreeMap<String, Value> = BTreeMap::new(); - for (key, value) in field { - match key.as_str() { - "type" | "name" | "doc" | "default" | "aliases" => continue, - _ => custom_attributes.insert(key.clone(), value.clone()), - }; - } - custom_attributes - } - /// Returns true if this `RecordField` is nullable, meaning the schema is a `UnionSchema` where the first variant is `Null`. pub fn is_nullable(&self) -> bool { match self.schema { diff --git a/avro/src/types.rs b/avro/src/types.rs index 124fe64..70759a0 100644 --- a/avro/src/types.rs +++ b/avro/src/types.rs @@ -813,7 +813,7 @@ impl Value { } Schema::Enum(EnumSchema { symbols, default, .. - }) => self.resolve_enum(symbols, default, field_default), + }) => self.resolve_enum(symbols, default.as_deref()), Schema::Array(inner) => { self.resolve_array(&inner.items, names, enclosing_namespace, depth) } @@ -1156,8 +1156,7 @@ impl Value { pub(crate) fn resolve_enum( self, symbols: &[String], - enum_default: &Option<String>, - _field_default: Option<&JsonValue>, + enum_default: Option<&str>, ) -> Result<Self, Error> { let validate_symbol = |symbol: String, symbols: &[String]| { if let Some(index) = symbols.iter().position(|item| item == &symbol) { @@ -1166,7 +1165,7 @@ impl Value { match enum_default { Some(default) => { if let Some(index) = symbols.iter().position(|item| item == default) { - Ok(Value::Enum(index as u32, default.clone())) + Ok(Value::Enum(index as u32, default.to_string())) } else { Err(Details::GetEnumDefault { symbol, @@ -1304,11 +1303,8 @@ impl Value { ref symbols, ref default, .. - }) => Value::try_from(value.clone())?.resolve_enum( - symbols, - default, - field.default.as_ref(), - )?, + }) => Value::try_from(value.clone())? + .resolve_enum(symbols, default.as_deref())?, Schema::Union(ref union_schema) => { let first = &union_schema.variants()[0]; // NOTE: this match exists only to optimize null defaults for large diff --git a/avro/src/util.rs b/avro/src/util.rs index 9fdd13b..35e0aac 100644 --- a/avro/src/util.rs +++ b/avro/src/util.rs @@ -50,35 +50,72 @@ pub const DEFAULT_SERDE_HUMAN_READABLE: bool = false; pub(crate) static SERDE_HUMAN_READABLE: OnceLock<bool> = OnceLock::new(); pub(crate) trait MapHelper { - fn string(&self, key: &str) -> Option<&str>; + fn string(&mut self, key: &'static str) -> AvroResult<Option<String>>; - fn name(&self) -> Option<&str> { - self.string("name") + fn str(&self, key: &'static str) -> AvroResult<Option<&str>>; + + fn name(&mut self) -> AvroResult<String> { + self.string("name")? + .ok_or_else(|| Details::GetNameField.into()) + } + + fn name_ref(&self) -> AvroResult<Option<&str>> { + self.str("name") } - fn doc(&self) -> Documentation { - self.string("doc").map(Into::into) + fn doc(&mut self) -> AvroResult<Documentation> { + self.string("doc") } - fn aliases(&self) -> Option<Vec<String>>; + fn aliases(&mut self) -> AvroResult<Option<Vec<String>>>; } impl MapHelper for Map<String, Value> { - fn string(&self, key: &str) -> Option<&str> { - self.get(key).and_then(|v| v.as_str()) + fn string(&mut self, key: &'static str) -> AvroResult<Option<String>> { + match self.remove(key) { + Some(Value::String(s)) => Ok(Some(s)), + Some(value) => Err(Details::GetStringInvalidType(key, value.description()).into()), + None => Ok(None), + } + } + + fn str(&self, key: &'static str) -> AvroResult<Option<&str>> { + match self.get(key) { + Some(Value::String(s)) => Ok(Some(s)), + Some(value) => Err(Details::GetStringInvalidType(key, value.description()).into()), + None => Ok(None), + } } - fn aliases(&self) -> Option<Vec<String>> { - // FIXME no warning when aliases aren't a json array of json strings - self.get("aliases") - .and_then(|aliases| aliases.as_array()) - .and_then(|aliases| { - aliases - .iter() - .map(|alias| alias.as_str()) - .map(|alias| alias.map(|a| a.to_string())) - .collect::<Option<_>>() - }) + fn aliases(&mut self) -> AvroResult<Option<Vec<String>>> { + match self.remove("aliases") { + Some(Value::Array(array)) => array + .into_iter() + .map(|v| match v { + Value::String(s) => Ok(s), + _ => Err(Details::GetAliasesFieldArrayInvalidType(v.description()).into()), + }) + .collect::<Result<Vec<_>, _>>() + .map(Some), + Some(value) => Err(Details::GetAliasesFieldInvalidType(value.description()).into()), + None => Ok(None), + } + } +} + +pub(crate) trait JsonValueDescriber { + fn description(&self) -> &'static str; +} +impl JsonValueDescriber for Value { + fn description(&self) -> &'static str { + match self { + Value::Null => "null", + Value::Bool(_) => "bool", + Value::Number(_) => "number", + Value::String(_) => "string", + Value::Array(_) => "array", + Value::Object(_) => "object", + } } }
