This is an automated email from the ASF dual-hosted git repository.
Kriskras99 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/avro-rs.git
The following commit(s) were added to refs/heads/main by this push:
new f48baf4 feat!: Make schema parser reuse the JSON allocations (#656)
f48baf4 is described below
commit f48baf472e9c8f4b7fddf12dcdc6f6616947b5db
Author: Kriskras99 <[email protected]>
AuthorDate: Mon Sep 7 21:45:57 2026 +0200
feat!: Make schema parser reuse the JSON allocations (#656)
* 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.
* fix build
* fix: Wrong error message and unnecessary clone
* fix: Small fixes and extra tests (from @JosephLenton)
---------
Co-authored-by: Martin Grigorov <[email protected]>
Co-authored-by: Martin Tzvetanov Grigorov <[email protected]>
---
avro/src/error.rs | 27 +++++
avro/src/reader/block.rs | 4 +-
avro/src/schema/mod.rs | 205 ++++++++++++++++++++++++++++++++----
avro/src/schema/name.rs | 22 ++--
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, 401 insertions(+), 216 deletions(-)
diff --git a/avro/src/error.rs b/avro/src/error.rs
index 6723340..0c05913 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 `namespace` 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 8858674..ad1687f 100644
--- a/avro/src/reader/block.rs
+++ b/avro/src/reader/block.rs
@@ -287,9 +287,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..b503f12 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 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") {
+ 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 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") {
+ 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,155 @@ 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(())
+ }
+
+ #[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/name.rs b/avro/src/schema/name.rs
index 057ae90..536e917 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};
@@ -110,14 +110,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 +257,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..37ef726 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).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(Details::GetFixedSizeFieldPositive(Value::Number(size))),
+ 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",
+ }
}
}
