From f6d4c0be99b0ac593cc05e27fbe7c495711990cf Mon Sep 17 00:00:00 2001 From: Kriskras99 Date: Fri, 4 Sep 2026 21:10:00 +0200 Subject: [PATCH 1/4] feat!: Make schema parser reuse the JSON allocations This changes the schema parser from iterating over the JSON into consuming it. This allows reusing the existing String allocations (and also collection allocations if the sizes match and the compiler can make it work). Because we remove everything from the JSON while consuming it, custom attributes is just everything leftover. Therefore this also fixes https://github.com/apache/avro-rs/issues/654 I've tried my best to improve the error messages where possible. This is a breaking change because the input for `Schema::parse` changes from a reference to an owned `Value`. Users can fix their usage by calling `.clone()` where needed. --- 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 67233400..311a6474 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 b73e01d9..73d045a1 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 fe3d72e1..fe58f070 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 { + pub fn parse(value: JsonValue) -> AvroResult { 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 { + pub(crate) fn parse_with_names(value: JsonValue, names: Names) -> AvroResult { 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 bfb69f2d..d290863a 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, + complex: &mut Map, enclosing_namespace: NamespaceRef, ) -> AvroResult { - 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 7f74af63..8b8f877e 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 { 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 { - 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 { - 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, + mut complex: Map, enclosing_namespace: NamespaceRef, ) -> AvroResult { // Try to parse this as a native complex type. fn parse_as_native_complex( - complex: &Map, + mut complex: Map, parser: &mut Parser, enclosing_namespace: NamespaceRef, ) -> AvroResult { - 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 { - 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, + mut complex: Map, enclosing_namespace: NamespaceRef, ) -> AvroResult { - 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 = 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::>() - })?; + 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::, _>>()?, + 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, - excluded: &[&'static str], - ) -> BTreeMap { - let mut custom_attributes: BTreeMap = 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) -> BTreeMap { + complex.into_iter().collect() } /// Parse a `serde_json::Value` representing a Avro enum type into a `Schema`. fn parse_enum( &mut self, - complex: &Map, + mut complex: Map, enclosing_namespace: NamespaceRef, ) -> AvroResult { - 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 = 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::>() - .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::, _>>(), + 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 = 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, + mut complex: Map, enclosing_namespace: NamespaceRef, ) -> AvroResult { 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, + mut complex: Map, enclosing_namespace: NamespaceRef, ) -> AvroResult { 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, enclosing_namespace: NamespaceRef, ) -> AvroResult { items - .iter() + .into_iter() .map(|v| self.parse(v, enclosing_namespace)) .collect::, _>>() .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, + mut complex: Map, enclosing_namespace: NamespaceRef, ) -> AvroResult { - 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 { 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 2bd6468a..38e08805 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, + mut field: Map, parser: &mut Parser, enclosing_record: &Name, ) -> AvroResult { - 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::>() - }) - }) - .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) -> BTreeMap { - let mut custom_attributes: BTreeMap = 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 124fe641..70759a09 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, - _field_default: Option<&JsonValue>, + enum_default: Option<&str>, ) -> Result { 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 9fdd13be..35e0aac6 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 = OnceLock::new(); pub(crate) trait MapHelper { - fn string(&self, key: &str) -> Option<&str>; + fn string(&mut self, key: &'static str) -> AvroResult>; - fn name(&self) -> Option<&str> { - self.string("name") + fn str(&self, key: &'static str) -> AvroResult>; + + fn name(&mut self) -> AvroResult { + self.string("name")? + .ok_or_else(|| Details::GetNameField.into()) + } + + fn name_ref(&self) -> AvroResult> { + self.str("name") } - fn doc(&self) -> Documentation { - self.string("doc").map(Into::into) + fn doc(&mut self) -> AvroResult { + self.string("doc") } - fn aliases(&self) -> Option>; + fn aliases(&mut self) -> AvroResult>>; } impl MapHelper for Map { - fn string(&self, key: &str) -> Option<&str> { - self.get(key).and_then(|v| v.as_str()) + fn string(&mut self, key: &'static str) -> AvroResult> { + 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> { + 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> { - // 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::>() - }) + fn aliases(&mut self) -> AvroResult>> { + 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::, _>>() + .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", + } } } From b4187d1e7b617a9653aa8c52e510849292d9e483 Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Sat, 5 Sep 2026 16:31:29 +0300 Subject: [PATCH 2/4] fix build --- avro/src/schema/name.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/avro/src/schema/name.rs b/avro/src/schema/name.rs index 6d28a8c2..536e9174 100644 --- a/avro/src/schema/name.rs +++ b/avro/src/schema/name.rs @@ -18,7 +18,7 @@ use crate::{ AvroResult, Error, Schema, error::Details, - util::MapHelper, + util::{JsonValueDescriber, MapHelper}, validator::{validate_namespace, validate_schema_name}, }; use serde::{Deserialize, Serialize, Serializer}; From 935953c35908a912e51bfd5d3f8a41f30609b69f Mon Sep 17 00:00:00 2001 From: Kriskras99 Date: Sun, 6 Sep 2026 14:17:40 +0200 Subject: [PATCH 3/4] fix: Wrong error message and unnecessary clone --- avro/src/error.rs | 2 +- avro/src/schema/parser.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/avro/src/error.rs b/avro/src/error.rs index 311a6474..0c059134 100644 --- a/avro/src/error.rs +++ b/avro/src/error.rs @@ -304,7 +304,7 @@ pub enum Details { #[error("No `name` field")] GetNameField, - #[error("Expected a string for the `name` field, got a {0}")] + #[error("Expected a string for the `namespace` field, got a {0}")] GetNamespaceFieldWrongType(&'static str), #[error("No `name` in record field")] diff --git a/avro/src/schema/parser.rs b/avro/src/schema/parser.rs index 8b8f877e..006737b5 100644 --- a/avro/src/schema/parser.rs +++ b/avro/src/schema/parser.rs @@ -740,7 +740,7 @@ impl Parser { let size = match size_opt { Some(Value::Number(size)) => size .as_u64() - .ok_or_else(|| Details::GetFixedSizeFieldPositive(Value::Number(size.clone()))), + .ok_or(Details::GetFixedSizeFieldPositive(Value::Number(size))), Some(v) => Err(Details::GetFixedSizeFieldInvalidType(v.description())), None => Err(Details::GetFixedSizeField), }?; From 2a938aefc3edeb5659ad1ecb2e250245968745f9 Mon Sep 17 00:00:00 2001 From: Kriskras99 Date: Mon, 7 Sep 2026 20:46:44 +0200 Subject: [PATCH 4/4] fix: Small fixes and extra tests (from @JosephLenton) --- avro/src/schema/mod.rs | 112 +++++++++++++++++++++++++++++++++++++- avro/src/schema/parser.rs | 2 +- 2 files changed, 111 insertions(+), 3 deletions(-) diff --git a/avro/src/schema/mod.rs b/avro/src/schema/mod.rs index fe58f070..b503f12c 100644 --- a/avro/src/schema/mod.rs +++ b/avro/src/schema/mod.rs @@ -546,7 +546,7 @@ 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 { - // Only clone the keys needed for the name parsing, can be a significant time/memory + // Only clone the values 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") { @@ -597,7 +597,7 @@ 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 { - // Only clone the keys needed for the name parsing, can be a significant time/memory + // Only clone the values 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") { @@ -5295,4 +5295,112 @@ mod tests { Ok(()) } + + #[test] + fn avro_rs_654_preserve_unknown_logical_type_on_outer_item() -> TestResult { + let raw_schema = r#"{ + "type": "array", + "logicalType": "blub", + "items": { + "type": "record", + "name": "k12_v13", + "fields": [ + { + "name": "key", + "type": "int", + "field-id": 12 + }, + { + "name": "value", + "type": "string", + "field-id": 13 + } + ] + } + }"#; + + let schema = Schema::parse_str(raw_schema)?; + + let output = serde_json::to_string_pretty(&schema).unwrap(); + pretty_assertions::assert_eq!( + r#"{ + "type": "array", + "items": { + "type": "record", + "name": "k12_v13", + "fields": [ + { + "name": "key", + "type": "int", + "field-id": 12 + }, + { + "name": "value", + "type": "string", + "field-id": 13 + } + ] + }, + "logicalType": "blub" +}"#, + output + ); + + let logical_type = schema.custom_attributes().unwrap().get("logicalType"); + assert_eq!( + logical_type, + Some(&serde_json::Value::String("blub".to_string())) + ); + + Ok(()) + } + + #[test] + fn avro_rs_654_preserve_unknown_logical_type_on_inner_item() -> TestResult { + let raw_schema = r#"{ + "type": "record", + "name": "test_record", + "fields": [ + { + "name": "example_map", + "type": { + "type": "array", + "logicalType": "fish", + "items": { + "type": "record", + "name": "k12_v13", + "fields": [ + { + "name": "key", + "type": "int", + "field-id": 12 + }, + { + "name": "value", + "type": "string", + "field-id": 13 + } + ] + } + } + } + ] + }"#; + + let schema = Schema::parse_str(raw_schema)?; + let Schema::Record(record) = &schema else { + panic!("Expected a record schema"); + }; + let example_map_schema = &record.fields[0].schema; + let logical_type = example_map_schema + .custom_attributes() + .unwrap() + .get("logicalType"); + assert_eq!( + logical_type, + Some(&serde_json::Value::String("fish".to_string())) + ); + + Ok(()) + } } diff --git a/avro/src/schema/parser.rs b/avro/src/schema/parser.rs index 006737b5..37ef726c 100644 --- a/avro/src/schema/parser.rs +++ b/avro/src/schema/parser.rs @@ -446,7 +446,7 @@ impl Parser { // 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).into()), _ => {} } match complex.remove("type") {