This is an automated email from the ASF dual-hosted git repository. Kriskras99 pushed a commit to branch feat/schema_parser_v2 in repository https://gitbox.apache.org/repos/asf/avro-rs.git
commit c10d80c4b86a94af5b07e1cd0c503e23ec1d3e02 Author: Kriskras99 <[email protected]> AuthorDate: Mon Sep 7 20:42:13 2026 +0200 wip --- Cargo.lock | 196 +++++++++++ avro/Cargo.toml | 5 + avro/benches/large_schema.avsc | 1 + avro/benches/schema_parsing.rs | 46 +++ avro/src/error.rs | 12 + avro/src/schema/mod.rs | 6 + avro/src/schema/parser2.rs | 745 +++++++++++++++++++++++++++++++++++++++++ 7 files changed, 1011 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 9f9d897..d593edc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,6 +8,19 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index"; checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "getrandom 0.3.4", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -74,6 +87,7 @@ dependencies = [ "serde_json", "sha2", "snap", + "sonic-rs", "strum", "thiserror", "uuid", @@ -183,6 +197,12 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index"; checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "bzip2" version = "0.6.1" @@ -461,6 +481,18 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index"; checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +[[package]] +name = "faststr" +version = "0.2.34" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "1ca7d44d22004409a61c393afb3369c8f7bb74abcae49fe249ee01dcc3002113" +dependencies = [ + "bytes", + "rkyv", + "serde", + "simdutf8", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -750,6 +782,26 @@ dependencies = [ "adler2", ] +[[package]] +name = "munge" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "5e17401f259eba956ca16491461b6e8f72913a0a114e39736ce404410f915a0c" +dependencies = [ + "munge_macro", +] + +[[package]] +name = "munge_macro" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "4568f25ccbd45ab5d5603dc34318c1ec56b117531781260002151b8530a9f931" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -929,6 +981,26 @@ dependencies = [ "unarray", ] +[[package]] +name = "ptr_meta" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "743da816b98c921cdbe8628ef7381b76f25ecf4da599fc80aca90eae7ef70cc0" +dependencies = [ + "ptr_meta_derive", +] + +[[package]] +name = "ptr_meta_derive" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "1c8d9ca532f185d5d4db7a7c9d51420b452168ea1c2b913953281bd6fe1fcbd0" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "quad-rand" version = "0.2.3" @@ -956,6 +1028,15 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index"; checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rancor" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "9b534442d0fcdb55d66f373d9cac6d33b6293a2335bc2136dbd06ce0e87d2572" +dependencies = [ + "ptr_meta", +] + [[package]] name = "rand" version = "0.9.4" @@ -1011,6 +1092,26 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "regex" version = "1.12.4" @@ -1052,6 +1153,41 @@ version = "1.9.3" source = "registry+https://github.com/rust-lang/crates.io-index"; checksum = "ba39f3699c378cd8970968dcbff9c43159ea4cfbd88d43c00b22f2ef10a435d2" +[[package]] +name = "rend" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "663ba70707f96e871406fe10d68128412e619b06d1d47cb91c3a4c6501176240" + +[[package]] +name = "rkyv" +version = "0.8.18" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "d9776093b7ca170454ab1406954f7b7d97a57c51dc6c0642957fb2ef25c2d399" +dependencies = [ + "bytes", + "hashbrown", + "indexmap", + "munge", + "ptr_meta", + "rancor", + "rend", + "rkyv_derive", + "tinyvec", + "uuid", +] + +[[package]] +name = "rkyv_derive" +version = "0.8.18" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "1c25ef604ac7dd839d44d64648952ea23c97866f124ff671b0ed2cf3ad9bb06e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.4", +] + [[package]] name = "rstest" version = "0.26.1" @@ -1187,6 +1323,12 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index"; checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + [[package]] name = "slab" version = "0.4.12" @@ -1199,6 +1341,45 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index"; checksum = "199905e6153d6405f9728fe44daace35f8f837bbf830bb6e85fbd5828709a886" +[[package]] +name = "sonic-number" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "3775c3390edf958191f1ab1e8c5c188907feebd0f3ce1604cb621f72961dbf32" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "sonic-rs" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "d971cc77a245ccf1756dbd1a87c3e7f709c0191464096510d43eec056d0f2c4f" +dependencies = [ + "ahash", + "bumpalo", + "bytes", + "cfg-if", + "faststr", + "itoa", + "ref-cast", + "serde", + "simdutf8", + "sonic-number", + "sonic-simd", + "thiserror", + "zmij", +] + +[[package]] +name = "sonic-simd" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "f99e664ecd2d85a68c87e3c7a3cfe691f647ea9e835de984aba4d54a41f817d4" +dependencies = [ + "cfg-if", +] + [[package]] name = "static_assertions" version = "1.1.0" @@ -1299,6 +1480,21 @@ dependencies = [ "serde_json", ] +[[package]] +name = "tinyvec" +version = "1.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "4cf0ded5c4e56918d8f8a339e1bb67d038d3bc6d144ac407904015ba2e4cde9b" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index"; +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "toml" version = "1.1.2+spec-1.1.0" diff --git a/avro/Cargo.toml b/avro/Cargo.toml index 7817d01..be061d5 100644 --- a/avro/Cargo.toml +++ b/avro/Cargo.toml @@ -47,6 +47,10 @@ name = "serde" harness = false name = "single" +[[bench]] +harness = false +name = "schema_parsing" + [dependencies] apache-avro-derive = { default-features = false, version = "0.23.0", path = "../avro_derive", optional = true } bigdecimal = { default-features = false, version = "0.4.10", features = ["std", "serde"] } @@ -68,6 +72,7 @@ thiserror = { default-features = false, version = "2.0.18" } uuid = { workspace = true } liblzma = { default-features = false, version = "0.4.6", optional = true } zstd = { default-features = false, version = "0.13.3", optional = true } +sonic-rs = "0.5.8" [target.'cfg(target_arch = "wasm32")'.dependencies] diff --git a/avro/benches/large_schema.avsc b/avro/benches/large_schema.avsc new file mode 100644 index 0000000..c5819b6 --- /dev/null +++ b/avro/benches/large_schema.avsc @@ -0,0 +1 @@ +["null",{"type":"record","name":"El","namespace":"foo.e.f.g.h.ac","fields":[{"name":"ref1","type":{"type":"record","name":"J","namespace":"foo.a.b.c.d","fields":[{"name":"attr11","type":"long","declared":"true"},{"name":"attr12","type":"long","declared":"true"}]}},{"name":"attr13","type":["null","string"],"default":null},{"name":"attr1","type":{"type":"long","date":"true"}},{"name":"attr2","type":["null","string"],"default":null},{"name":"attr3","type":["null","string"],"default":null},{ [...] \ No newline at end of file diff --git a/avro/benches/schema_parsing.rs b/avro/benches/schema_parsing.rs new file mode 100644 index 0000000..c2ac53f --- /dev/null +++ b/avro/benches/schema_parsing.rs @@ -0,0 +1,46 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use apache_avro::schema::Schema; +use criterion::{Criterion, criterion_group, criterion_main}; +use std::time::Duration; + +fn read_large_record() -> String { + std::fs::read_to_string("benches/large_schema.avsc").unwrap() +} + +fn bench_large_record(c: &mut Criterion) { + let string = read_large_record(); + c.bench_function("large record", |b| { + b.iter(|| Schema::parse_str(&string)); + }); +} +fn bench_large_record2(c: &mut Criterion) { + let string = read_large_record(); + c.bench_function("large record", |b| { + b.iter(|| Schema::parse_str2(&string)); + }); +} + +criterion_group!( + name = benches; + config = Criterion::default().sample_size(200).measurement_time(Duration::from_secs(10)); + targets = + bench_large_record, + bench_large_record2, +); +criterion_main!(benches); diff --git a/avro/src/error.rs b/avro/src/error.rs index 58f9145..f20eba2 100644 --- a/avro/src/error.rs +++ b/avro/src/error.rs @@ -428,9 +428,15 @@ pub enum Details { #[error("logicalType must be a string, but is {0:?}")] GetLogicalTypeFieldType(serde_json::Value), + #[error("logicalType must be a string, but is {0:?}")] + GetLogicalTypeFieldType2(sonic_rs::JsonType), + #[error("Unknown complex type: {0}")] GetComplexType(serde_json::Value), + #[error("Unknown complex type: {0:?}")] + GetComplexType2(sonic_rs::JsonType), + #[error("No `type` in complex type")] GetComplexTypeField, @@ -729,6 +735,12 @@ pub enum Details { position: usize, total_elements: usize, }, + + #[error("{0}")] + Sonic(sonic_rs::Error), + + #[error("{0}")] + Custom(String), } #[derive(thiserror::Error, PartialEq)] diff --git a/avro/src/schema/mod.rs b/avro/src/schema/mod.rs index b503f12..ca5d471 100644 --- a/avro/src/schema/mod.rs +++ b/avro/src/schema/mod.rs @@ -20,10 +20,12 @@ mod builders; mod name; mod parser; +mod parser2; mod record; mod resolve; mod union; +use crate::schema::parser2::Parser2; pub(crate) use crate::schema::resolve::{ ResolvedOwnedSchema, resolve_names, resolve_names_with_schemata, }; @@ -530,6 +532,10 @@ impl Schema { parser.parse_str(input) } + pub fn parse_str2(input: &str) -> Result<Schema, Error> { + Parser2::parse_str(input) + } + /// Create an array of `Schema`'s from a list of named JSON Avro schemas (Record, Enum, and /// Fixed). /// diff --git a/avro/src/schema/parser2.rs b/avro/src/schema/parser2.rs new file mode 100644 index 0000000..38c6fd8 --- /dev/null +++ b/avro/src/schema/parser2.rs @@ -0,0 +1,745 @@ +use crate::error::Details; +use crate::schema::{ + Alias, ArraySchema, DecimalSchema, EnumSchema, FixedSchema, InnerDecimalSchema, MapSchema, + Name, NamespaceRef, RecordField, RecordSchema, UnionSchema, UuidSchema, +}; +use crate::{AvroResult, Error, Schema}; +use log::warn; +use sonic_rs::{ArrayJsonIter, JsonType, JsonValueTrait, LazyValue}; +use std::borrow::Cow; +use std::collections::{BTreeMap, HashSet}; +use std::ops::Deref; +use std::sync::LazyLock; + +static BYTES_UUID_WARNING: LazyLock<()> = + LazyLock::new(|| warn!("`bytes` with a logical type of `uuid` is deprecated")); + +#[derive(Default)] +pub struct Parser2 { + known_names: HashSet<Name>, + unknown_references: HashSet<Name>, +} + +impl Parser2 { + /// Create a `Schema` from a string representing a JSON Avro schema. + pub(super) fn parse_str(input: &str) -> AvroResult<Schema> { + let lazy = sonic_rs::get_from_str::<[usize; 0]>(input, []).map_err(Details::Sonic)?; + let mut parser = Self::default(); + let schema = parser.parse(lazy, None)?; + if parser.has_unknown_references() { + Err(Details::SchemaResolutionError( + parser.unknown_references.into_iter().next().unwrap(), + ) + .into()) + } else { + Ok(schema) + } + } + + fn has_unknown_references(&mut self) -> bool { + self.unknown_references + .retain(|r| !self.known_names.contains(r)); + !self.unknown_references.is_empty() + } + + fn add_reference(&mut self, name: &Name) { + if !self.known_names.contains(name) { + self.unknown_references.insert(name.clone()); + } + } + + fn parse(&mut self, value: LazyValue, enclosing_namespace: NamespaceRef) -> AvroResult<Schema> { + match value.get_type() { + JsonType::String => { + self.parse_string(value.as_str().expect("unreachable"), enclosing_namespace) + } + JsonType::Object => self.parse_complex(value, enclosing_namespace), + JsonType::Array => self.parse_union( + value.into_array_iter().expect("unreachable"), + enclosing_namespace, + ), + _ => Err(Details::ParseSchemaFromValidJson.into()), + } + } + + fn parse_string( + &mut self, + value: &str, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + match value { + "null" => Ok(Schema::Null), + "boolean" => Ok(Schema::Boolean), + "int" => Ok(Schema::Int), + "long" => Ok(Schema::Long), + "double" => Ok(Schema::Double), + "float" => Ok(Schema::Float), + "bytes" => Ok(Schema::Bytes), + "string" => Ok(Schema::String), + _ => { + let name = Name::new_with_enclosing_namespace(value, enclosing_namespace)?; + self.unknown_references.insert(name.clone()); + Ok(Schema::Ref { name }) + } + } + } + + fn parse_union( + &mut self, + array: ArrayJsonIter, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let mut builder = UnionSchema::builder(); + for value in array { + let value = value.map_err(Details::Sonic)?; + let schema = self.parse(value, enclosing_namespace)?; + builder.variant(schema)?; + } + Ok(Schema::Union(builder.build())) + } + + fn parse_complex( + &mut self, + value: LazyValue, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + assert!(value.is_object()); + + let ty_value = value + .pointer(["type"]) + .ok_or(Details::GetComplexTypeField)?; + let ty = ty_value + .as_str() + .ok_or_else(|| Details::GetComplexType2(ty_value.get_type()))?; + let logical_ty_value = value.pointer(["logicalType"]); + let logical_ty = if let Some(val) = &logical_ty_value { + Some( + val.as_str() + .ok_or_else(|| Details::GetLogicalTypeFieldType2(val.get_type()))?, + ) + } else { + None + }; + + // This clone is practically a copy + let object_iter = value + .clone() + .into_object_iter() + .expect("This is an object") + .filter_map(|r| { + if let Ok((key, _)) = &r + && key == "type" + { + None + } else { + Some(r.map_err(|e| Error::new(Details::Sonic(e)))) + } + }); + match (ty, logical_ty) { + ("null", _) => Ok(Schema::Null), + ("boolean", _) => Ok(Schema::Boolean), + ("int", Some("date")) => Ok(Schema::Date), + ("int", Some("time-millis")) => Ok(Schema::TimeMillis), + ("int", _) => Ok(Schema::Int), + ("long", Some("time-micros")) => Ok(Schema::TimeMicros), + ("long", Some("timestamp-millis")) => Ok(Schema::TimestampMillis), + ("long", Some("timestamp-micros")) => Ok(Schema::TimestampMicros), + ("long", Some("timestamp-nanos")) => Ok(Schema::TimestampNanos), + ("long", Some("local-timestamp-millis")) => Ok(Schema::LocalTimestampMillis), + ("long", Some("local-timestamp-micros")) => Ok(Schema::LocalTimestampMicros), + ("long", Some("local-timestamp-nanos")) => Ok(Schema::LocalTimestampNanos), + ("long", _) => Ok(Schema::Long), + ("float", _) => Ok(Schema::Float), + ("double", _) => Ok(Schema::Double), + ("bytes", Some("decimal")) => { + self.parse_decimal(object_iter, "bytes", enclosing_namespace) + } + ("bytes", Some("big-decimal")) => Ok(Schema::BigDecimal), + ("bytes", Some("uuid")) => { + let _ = BYTES_UUID_WARNING.deref(); + Ok(Schema::Uuid(UuidSchema::Bytes)) + } + ("bytes", _) => Ok(Schema::Bytes), + ("string", Some("uuid")) => Ok(Schema::Uuid(UuidSchema::String)), + ("string", _) => Ok(Schema::String), + ("record", _) => self.parse_record(object_iter, enclosing_namespace), + ("enum", _) => self.parse_enum(object_iter, enclosing_namespace), + ("array", _) => self.parse_array(object_iter, enclosing_namespace), + ("maps", _) => self.parse_map(object_iter, enclosing_namespace), + ("fixed", Some("decimal")) => { + self.parse_decimal(object_iter, "fixed", enclosing_namespace) + } + ("fixed", Some("uuid")) => self.parse_uuid_fixed(object_iter, enclosing_namespace), + ("fixed", Some("duration")) => self.parse_duration(object_iter, enclosing_namespace), + ("fixed", _) => self.parse_fixed(object_iter, enclosing_namespace), + (_, None) => { + let name = Name::new_with_enclosing_namespace(ty, enclosing_namespace)?; + self.add_reference(&name); + Ok(Schema::Ref { name }) + } + (_, Some(lt)) => { + panic!( + "References cannot be combined with logical types, reference: {ty}, logical type: {lt}" + ); + } + } + } + + fn parse_decimal<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + ty: &str, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let mut precision = None; + let mut scale = None; + + let extracted = + object.filter_map(|r| match r.as_ref().map(|(k, v)| (k.as_ref(), v)) { + Ok(("precision", value)) if value.is_u64() => { + let value = value.as_u64().expect("Is a u64"); + match usize::try_from(value) { + Ok(0) => { + Some(Err(Details::Custom("The `precision` for decimal cannot be 0".into()).into())) + } + Ok(value) => { + precision = Some(value); + None + } + Err(err) => Some(Err(Details::ConvertU64ToUsize(err, value).into())) + } + } + Ok(("precision", value)) if value.is_i64() => Some(Err(Details::Custom("Expected a non-zero positive integer for decimal `precision` got a negative number".into()).into())), + Ok(("precision", value)) if value.is_f64() => Some(Err(Details::Custom("Expected a non-zero positive integer for decimal `precision` got a floating point number".into()).into())), + Ok(("precision", value)) => Some(Err(Details::Custom(format!("Expected a non-zero positive integer for decimal `precision` got {:?}", value.get_type())).into())), + Ok(("scale", value)) if value.is_u64() => { + let value = value.as_u64().expect("Is a u64"); + match usize::try_from(value) { + Ok(value) => { + scale = Some(value); + None + } + Err(err) => Some(Err(Details::ConvertU64ToUsize(err, value).into())) + } + } + Ok(("scale", value)) if value.is_i64() => Some(Err(Details::Custom("Expected a positive integer for decimal `scale` got a negative number".into()).into())), + Ok(("scale", value)) if value.is_f64() => Some(Err(Details::Custom("Expected a positive integer for decimal `scale` got a floating point number".into()).into())), + Ok(("scale", value)) => Some(Err(Details::Custom(format!("Expected a positive integer for decimal `scale` got {:?}", value.get_type())).into())), + _ => Some(r), + }); + + match ty { + "bytes" => { + for result in extracted { + result?; + } + let precision = precision + .ok_or_else(|| Details::Custom("Missing `precision` for decimal".into()))?; + let scale = scale.unwrap_or(0); + if scale > precision { + return Err(Details::DecimalPrecisionLessThanScale { scale, precision }.into()); + } + + Ok(Schema::Decimal(DecimalSchema { + precision, + scale, + inner: InnerDecimalSchema::Bytes, + })) + } + "fixed" => { + let fixed = self.parse_fixed_bare(extracted, enclosing_namespace)?; + let precision = precision + .ok_or_else(|| Details::Custom("Missing `precision` for decimal".into()))?; + let scale = scale.unwrap_or(0); + if scale > precision { + return Err(Details::DecimalPrecisionLessThanScale { scale, precision }.into()); + } + + let max_precision = (2usize.pow((8 * (fixed.size - 1)) as u32) - 1).ilog10(); + if (max_precision as usize) < precision { + return Err(Details::Custom(format!("Maximum precision for Fixed({}) is {max_precision} but precision is {precision}", fixed.size)).into()); + } + + Ok(Schema::Decimal(DecimalSchema { + precision, + scale, + inner: InnerDecimalSchema::Fixed(fixed), + })) + } + _ => unreachable!(), + } + } + + fn parse_record<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let mut name = None; + let mut namespace = None; + let mut aliases = None; + let mut doc = None; + let mut fields = None; + let mut attributes = BTreeMap::new(); + + for result in object { + let (key, value) = result?; + // TODO: panics to actual errors + match (key.as_ref(), value.get_type()) { + ("name", _) if name.is_some() => panic!("Duplicate `name` field"), + ("name", JsonType::String) => { + name = Some(value.as_str().expect("Is a string").to_string()) + } + ("name", ty) => panic!("`name` must be a string not a {ty:?}"), + ("namespace", _) if namespace.is_some() => panic!("Duplicate `namespace` field"), + ("namespace", JsonType::String) => { + namespace = Some(value.as_str().expect("Is a string").to_string()) + } + ("namespace", ty) => panic!("`namespace` must be a string not a {ty:?}"), + ("aliases", _) if aliases.is_none() => panic!("Duplicate `aliases` field"), + ("aliases", JsonType::Array) => { + aliases = Some( + self.parse_aliases(value.into_array_iter().unwrap(), enclosing_namespace)?, + ) + } + ("aliases", ty) => panic!("`aliases` must be a string not a {ty:?}"), + ("doc", _) if doc.is_some() => panic!("Duplicate `doc` field"), + ("doc", JsonType::String) => { + doc = Some(value.as_str().expect("Is a string").to_string()) + } + ("doc", ty) => panic!("`doc` must be a string not a {ty:?}"), + ("fields", _) if fields.is_some() => panic!("Duplicate `fields` field"), + ("fields", JsonType::Array) => { + fields = Some(self.parse_fields( + value.into_array_iter().expect("Is an array"), + enclosing_namespace, + )) + } + ("fields", ty) => panic!("`fields` must be an positive integer not a {ty:?}"), + (_, _) => { + if attributes.contains_key(key.as_ref()) { + panic!("Duplicate custom attribute found for {key}"); + } + attributes.insert(key.into_owned(), sonic_value_to_serde_value(value)); + } + } + } + + let Some(name) = name else { + return Err(Details::GetNameField.into()); + }; + let Some(fields) = fields else { + panic!("`fields` is missing on the struct") + }; + let name = + Name::new_with_enclosing_namespace(name, namespace.as_deref().or(enclosing_namespace))?; + + Ok(Schema::Record( + RecordSchema::builder() + .name(name) + .aliases(aliases) + .doc(doc) + .fields(fields?) + .attributes(attributes) + .build(), + )) + } + + fn parse_enum<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let mut name = None; + let mut namespace = None; + let mut aliases = None; + let mut doc = None; + let mut symbols = None; + let mut default = None; + let mut attributes = BTreeMap::new(); + + for result in object { + let (key, value) = result?; + // TODO: panics to actual errors + match (key.as_ref(), value.get_type()) { + ("name", _) if name.is_some() => panic!("Duplicate `name` field"), + ("name", JsonType::String) => { + name = Some(value.as_str().expect("Is a string").to_string()) + } + ("name", ty) => panic!("`name` must be a string not a {ty:?}"), + ("namespace", _) if namespace.is_some() => panic!("Duplicate `namespace` field"), + ("namespace", JsonType::String) => { + namespace = Some(value.as_str().expect("Is a string").to_string()) + } + ("namespace", ty) => panic!("`namespace` must be a string not a {ty:?}"), + ("aliases", _) if aliases.is_none() => panic!("Duplicate `aliases` field"), + ("aliases", JsonType::Array) => { + aliases = Some( + self.parse_aliases(value.into_array_iter().unwrap(), enclosing_namespace)?, + ) + } + ("aliases", ty) => panic!("`aliases` must be a string not a {ty:?}"), + ("doc", _) if doc.is_some() => panic!("Duplicate `doc` field"), + ("doc", JsonType::String) => { + doc = Some(value.as_str().expect("Is a string").to_string()) + } + ("doc", ty) => panic!("`doc` must be a string not a {ty:?}"), + ("symbols", _) if symbols.is_some() => panic!("Duplicate `symbols` field"), + ("symbols", JsonType::Array) => { + symbols = Some( + value.into_array_iter().unwrap().map(|r| { + r.map_err(Details::Sonic).and_then(|v| v.as_str().map(str::to_string).ok_or_else(|| Details::Custom(format!("`symbols` must be an array of string, but got an {:?} in the array", v.get_type())))) + }).collect::<Result<Vec<_>, _>>()?, + ) + } + ("default", _) if default.is_some() => panic!("Duplicate `default` field"), + ("default", JsonType::String) => default = Some(value.as_str().expect("Is a string").to_string()), + ("default", ty) => panic!("`default` must be a string not a {ty:?}"), + (_, _) => { + if attributes.contains_key(key.as_ref()) { + panic!("Duplicate custom attribute found for {key}"); + } + attributes.insert(key.into_owned(), sonic_value_to_serde_value(value)); + } + } + } + + let Some(name) = name else { + return Err(Details::GetNameField.into()); + }; + let Some(symbols) = symbols else { + panic!("`symbols` is missing on the struct") + }; + let name = + Name::new_with_enclosing_namespace(name, namespace.as_deref().or(enclosing_namespace))?; + if let Some(default) = &default + && !symbols.contains(default) + { + panic!("`default` of {default} does not exist in `symbols`") + } + + Ok(Schema::Enum(EnumSchema { + name, + aliases, + doc, + symbols, + default, + attributes, + })) + } + + fn parse_array<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let mut items = None; + let mut attributes = BTreeMap::new(); + + for result in object { + let (key, value) = result?; + match key.as_ref() { + "items" if items.is_some() => panic!("Duplicate `items` field"), + "items" => items = Some(self.parse(value, enclosing_namespace)?), + _ => { + if attributes.contains_key(key.as_ref()) { + panic!("Duplicate custom attribute found for {key}"); + } + attributes.insert(key.into_owned(), sonic_value_to_serde_value(value)); + } + } + } + + let Some(items) = items else { + panic!("Mising `items` for array") + }; + + Ok(Schema::Array(ArraySchema { + items: Box::new(items), + attributes, + })) + } + + fn parse_map<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let mut ty = None; + let mut attributes = BTreeMap::new(); + + for result in object { + let (key, value) = result?; + match key.as_ref() { + "type" if ty.is_some() => panic!("Duplicate `type` field"), + "type" => ty = Some(self.parse(value, enclosing_namespace)?), + _ => { + if attributes.contains_key(key.as_ref()) { + panic!("Duplicate custom attribute found for {key}"); + } + attributes.insert(key.into_owned(), sonic_value_to_serde_value(value)); + } + } + } + + let Some(ty) = ty else { + panic!("Mising `items` for array") + }; + + Ok(Schema::Map(MapSchema { + types: Box::new(ty), + attributes, + })) + } + + fn parse_uuid_fixed<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let filtered = object.filter(|r| { + if let Ok((key, _)) = r + && key == "logicalType" + { + false + } else { + true + } + }); + let fixed = self.parse_fixed_bare(filtered, enclosing_namespace)?; + if fixed.size != 16 { + // TODO: Wrong error? + Err(Details::ConvertFixedToUuid(fixed.size).into()) + } else { + Ok(Schema::Uuid(UuidSchema::Fixed(fixed))) + } + } + + fn parse_duration<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + let filtered = object.filter(|r| { + if let Ok((key, _)) = r + && key == "logicalType" + { + false + } else { + true + } + }); + Ok(Schema::Duration( + self.parse_fixed_bare(filtered, enclosing_namespace)?, + )) + } + + fn parse_fixed<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Schema> { + self.parse_fixed_bare(object, enclosing_namespace) + .map(|f| Schema::Uuid(UuidSchema::Fixed(f))) + } + + fn parse_fixed_bare<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<FixedSchema> { + let mut name = None; + let mut namespace = None; + let mut aliases = None; + let mut doc = None; + let mut size = None; + let mut attributes = BTreeMap::new(); + + for result in object { + let (key, value) = result?; + // TODO: panics to actual errors + match (key.as_ref(), value.get_type()) { + ("name", _) if name.is_some() => panic!("Duplicate `name` field"), + ("name", JsonType::String) => { + name = Some(value.as_str().expect("Is a string").to_string()) + } + ("name", ty) => panic!("`name` must be a string not a {ty:?}"), + ("namespace", _) if namespace.is_some() => panic!("Duplicate `namespace` field"), + ("namespace", JsonType::String) => { + namespace = Some(value.as_str().expect("Is a string").to_string()) + } + ("namespace", ty) => panic!("`namespace` must be a string not a {ty:?}"), + ("aliases", _) if aliases.is_none() => panic!("Duplicate `aliases` field"), + ("aliases", JsonType::Array) => { + aliases = Some( + self.parse_aliases(value.into_array_iter().unwrap(), enclosing_namespace)?, + ) + } + ("aliases", ty) => panic!("`aliases` must be a string not a {ty:?}"), + ("doc", _) if doc.is_some() => panic!("Duplicate `doc` field"), + ("doc", JsonType::String) => { + doc = Some(value.as_str().expect("Is a string").to_string()) + } + ("doc", ty) => panic!("`doc` must be a string not a {ty:?}"), + ("size", _) if size.is_some() => panic!("Duplicate `size` field"), + ("size", _) if value.is_u64() => size = Some(value.as_u64().expect("Is a u64")), + ("size", _) if value.is_i64() => panic!("`size` must be an positive integer"), + ("size", _) if value.is_f64() => { + panic!("`size` must be an positive integer not a float") + } + ("size", ty) => panic!("`size` must be an positive integer not a {ty:?}"), + (_, _) => { + if attributes.contains_key(key.as_ref()) { + panic!("Duplicate custom attribute found for {key}"); + } + attributes.insert(key.into_owned(), sonic_value_to_serde_value(value)); + } + } + } + + let Some(name) = name else { + return Err(Details::GetNameField.into()); + }; + let Some(size) = size else { + return Err(Details::GetFixedSizeField.into()); + }; + let name = + Name::new_with_enclosing_namespace(name, namespace.as_deref().or(enclosing_namespace))?; + + Ok(FixedSchema { + name, + aliases: None, + doc, + size: usize::try_from(size).map_err(|e| Details::ConvertU64ToUsize(e, size))?, + attributes, + }) + } + + fn parse_aliases( + &mut self, + array: ArrayJsonIter, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Vec<Alias>> { + // This function is only entered if `aliases` exist, and the most common length of `aliases` + // is probably 1. + let mut aliases = Vec::with_capacity(1); + + for result in array { + let value = result.map_err(Details::Sonic)?; + if let Some(str) = value.as_str() { + aliases.push(Alias::new_with_enclosing_namespace( + str, + enclosing_namespace, + )?); + } else { + panic!( + "aliases must be an array of strings, found an {:?} in the array", + value.get_type() + ) + } + } + + aliases.shrink_to_fit(); + Ok(aliases) + } + + fn parse_fields( + &mut self, + array: ArrayJsonIter, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<Vec<RecordField>> { + let mut fields = Vec::new(); + for result in array { + let item = result.map_err(Details::Sonic)?; + let ty = item.get_type(); + if let Some(object) = item.into_object_iter() { + let field = self.parse_field( + object.map(|r| r.map_err(|e| Error::new(Details::Sonic(e)))), + enclosing_namespace, + )?; + fields.push(field); + } else { + panic!("Expected array of objects for `fields` but found {ty:?} in the array") + } + } + Ok(fields) + } + + fn parse_field<'de>( + &mut self, + object: impl Iterator<Item = AvroResult<(Cow<'de, str>, LazyValue<'de>)>>, + enclosing_namespace: NamespaceRef, + ) -> AvroResult<RecordField> { + let mut name = None; + let mut doc = None; + let mut ty = None; + let mut order_found = false; + let mut aliases = None; + let mut default = None; + let mut attributes = BTreeMap::new(); + + for result in object { + let (key, value) = result?; + // TODO: panics to actual errors + match (key.as_ref(), value.get_type()) { + ("name", _) if name.is_some() => panic!("Duplicate `name` field"), + ("name", JsonType::String) => { + name = Some(value.as_str().expect("Is a string").to_string()) + } + ("name", ty) => panic!("`name` must be a string not a {ty:?}"), + ("doc", _) if doc.is_some() => panic!("Duplicate `doc` field"), + ("doc", JsonType::String) => { + doc = Some(value.as_str().expect("Is a string").to_string()) + } + ("doc", ty) => panic!("`doc` must be a string not a {ty:?}"), + ("type", _) if ty.is_some() => panic!("Duplicate `type` field"), + ("type", _) => ty = Some(self.parse(value, enclosing_namespace)), + ("order", _) if order_found => panic!("Duplicate `order` field"), + ("order", JsonType::String) => match value.as_str().expect("Is a string") { + "ascending" | "descending" | "ignore" => order_found = true, + v => panic!("Unexpected value for `order`: {v}"), + } + ("order", ty) => panic!("`order` must be a string not a {ty:?}"), + ("aliases", _) if aliases.is_none() => panic!("Duplicate `aliases` field"), + ("aliases", JsonType::Array) => { + aliases = Some( + value.into_array_iter().unwrap().map(|r| { + r.map_err(Details::Sonic).and_then(|v| v.as_str().map(str::to_string).ok_or_else(|| Details::Custom(format!("`aliases` must be an array of string, but got an {:?} in the array", v.get_type())))) + }).collect::<Result<Vec<_>, _>>()?, + ) + } + ("aliases", ty) => panic!("`aliases` must be a string not a {ty:?}"), + ("default", _) if default.is_some() => panic!("Duplicate `default` field"), + ("default", _) => { + default = Some(serde_json::from_str(value.as_raw_str()).expect("This is valid JSON")) + } + (_, _) => { + if attributes.contains_key(key.as_ref()) { + panic!("Duplicate custom attribute found for {key}"); + } + attributes.insert(key.into_owned(), sonic_value_to_serde_value(value)); + } + } + } + + let Some(name) = name else { + panic!("`name` is missing for field") + }; + let Some(ty) = ty else { + panic!("`ty` is missing for field") + }; + + Ok(RecordField::builder() + .name(name) + .doc(doc) + .maybe_aliases(aliases) + .maybe_default(default) + .schema(ty?) + .custom_attributes(attributes) + .build()) + } +} + +fn sonic_value_to_serde_value(value: LazyValue) -> serde_json::Value { + serde_json::from_str(value.as_raw_str()).expect("This should parse") +}
