You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@avro.apache.org by kr...@apache.org on 2026/03/19 12:56:02 UTC
(avro-rs) 02/05: feat: `SchemaAwareDeserializer` (no tests)
This is an automated email from the ASF dual-hosted git repository.
kriskras99 pushed a commit to branch feat/full_enum_support
in repository https://gitbox.apache.org/repos/asf/avro-rs.git
commit 764f119de4a67de855dffe5883eae1bcdc58953e
Author: Kriskras99 <gi...@kriskras99.nl>
AuthorDate: Fri Mar 13 18:50:41 2026 +0100
feat: `SchemaAwareDeserializer` (no tests)
---
avro/src/error.rs | 11 +-
avro/src/schema/union.rs | 13 +-
avro/src/serde/deser_schema/array.rs | 99 ++++
avro/src/serde/deser_schema/enums.rs | 233 ++++++++++
avro/src/serde/deser_schema/identifier.rs | 269 +++++++++++
avro/src/serde/deser_schema/map.rs | 147 ++++++
avro/src/serde/deser_schema/mod.rs | 741 ++++++++++++++++++++++++++++++
avro/src/serde/deser_schema/record.rs | 93 ++++
avro/src/serde/deser_schema/tuple.rs | 123 +++++
avro/src/serde/mod.rs | 1 +
10 files changed, 1728 insertions(+), 2 deletions(-)
diff --git a/avro/src/error.rs b/avro/src/error.rs
index 67ff13f..123bf3c 100644
--- a/avro/src/error.rs
+++ b/avro/src/error.rs
@@ -156,7 +156,6 @@ pub enum Details {
#[error("Union index {index} out of bounds: {num_variants}")]
GetUnionVariant { index: i64, num_variants: usize },
- #[deprecated(since = "0.20.0", note = "This error variant is not generated anymore")]
#[error("Enum symbol index out of bounds: {num_variants}")]
EnumSymbolIndex { index: usize, num_variants: usize },
@@ -579,6 +578,16 @@ pub enum Details {
#[error("Failed to deserialize Avro value into value: {0}")]
DeserializeValue(String),
+ #[error("Failed to deserialize value of type {value_type} using schema {schema:?}: {value}")]
+ DeserializeSchemaAware {
+ value_type: &'static str,
+ value: String,
+ schema: Schema,
+ },
+
+ #[error("Only expected `deserialize_identifier` to be called but `{0}` was called")]
+ DeserializeIdentifier(&'static str),
+
#[error("Failed to write buffer bytes during flush: {0}")]
WriteBytes(#[source] std::io::Error),
diff --git a/avro/src/schema/union.rs b/avro/src/schema/union.rs
index 31affe7..c8bef75 100644
--- a/avro/src/schema/union.rs
+++ b/avro/src/schema/union.rs
@@ -74,7 +74,18 @@ impl UnionSchema {
&self.schemas
}
- /// Returns true if the any of the variants of this `UnionSchema` is `Null`.
+ /// Get the variant at the given index.
+ pub fn get_variant(&self, index: usize) -> Result<&Schema, Error> {
+ self.schemas.get(index).ok_or_else(|| {
+ Details::GetUnionVariant {
+ index: index as i64,
+ num_variants: self.schemas.len(),
+ }
+ .into()
+ })
+ }
+
+ /// Returns true if any of the variants of this `UnionSchema` is `Null`.
pub fn is_nullable(&self) -> bool {
self.variant_index.contains_key(&SchemaKind::Null)
}
diff --git a/avro/src/serde/deser_schema/array.rs b/avro/src/serde/deser_schema/array.rs
new file mode 100644
index 0000000..c1da742
--- /dev/null
+++ b/avro/src/serde/deser_schema/array.rs
@@ -0,0 +1,99 @@
+// 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 super::{Config, SchemaAwareDeserializer};
+use crate::schema::ArraySchema;
+use crate::util::{zag_i32, zag_i64};
+use crate::{Error, Schema};
+use serde::de::{DeserializeSeed, SeqAccess};
+use std::borrow::Borrow;
+use std::io::Read;
+
+/// Deserialize sequences from an Avro array.
+pub struct ArrayDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
+ reader: &'r mut R,
+ schema: &'s Schema,
+ config: Config<'s, S>,
+ /// Track where we are in reading the array
+ remaining: Option<u32>,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> ArrayDeserializer<'s, 'r, R, S> {
+ pub fn new(
+ reader: &'r mut R,
+ schema: &'s ArraySchema,
+ config: Config<'s, S>,
+ ) -> Result<Self, Error> {
+ let schema = if let Schema::Ref { name } = schema.items.as_ref() {
+ config.get_schema(name)?
+ } else {
+ &schema.items
+ };
+ let remaining = Self::read_block_header(reader)?;
+ Ok(Self {
+ reader,
+ schema,
+ config,
+ remaining,
+ })
+ }
+
+ fn read_block_header(reader: &mut R) -> Result<Option<u32>, Error> {
+ let remaining = zag_i32(reader)?;
+ if remaining < 0 {
+ // If the block size is negative the next number is the size of the block in bytes
+ let _bytes = zag_i64(reader)?;
+ }
+ if remaining == 0 {
+ // If the block size is zero the array is finished
+ Ok(None)
+ } else {
+ Ok(Some(remaining.unsigned_abs()))
+ }
+ }
+}
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> SeqAccess<'de> for ArrayDeserializer<'s, 'r, R, S> {
+ type Error = Error;
+
+ fn next_element_seed<T>(&mut self, seed: T) -> Result<Option<T::Value>, Self::Error>
+ where
+ T: DeserializeSeed<'de>,
+ {
+ if let Some(mut remaining) = self.remaining {
+ let value = seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ self.schema,
+ self.config,
+ )?)?;
+ remaining -= 1;
+ if remaining == 0 {
+ self.remaining = Self::read_block_header(self.reader)?;
+ } else {
+ self.remaining = Some(remaining);
+ }
+ Ok(Some(value))
+ } else {
+ Ok(None)
+ }
+ }
+
+ fn size_hint(&self) -> Option<usize> {
+ self.remaining
+ .map(|x| usize::try_from(x).unwrap_or(usize::MAX))
+ }
+}
diff --git a/avro/src/serde/deser_schema/enums.rs b/avro/src/serde/deser_schema/enums.rs
new file mode 100644
index 0000000..bb2e7c5
--- /dev/null
+++ b/avro/src/serde/deser_schema/enums.rs
@@ -0,0 +1,233 @@
+// 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 super::{Config, DESERIALIZE_ANY, SchemaAwareDeserializer, identifier::IdentifierDeserializer};
+use crate::error::Details;
+use crate::schema::{EnumSchema, UnionSchema};
+use crate::util::zag_i32;
+use crate::{Error, Schema};
+use serde::Deserializer;
+use serde::de::{DeserializeSeed, EnumAccess, Unexpected, VariantAccess, Visitor};
+use std::borrow::Borrow;
+use std::io::Read;
+
+/// Deserializer for plain enums.
+pub struct PlainEnumDeserializer<'s, 'r, R: Read> {
+ reader: &'r mut R,
+ symbols: &'s [String],
+}
+
+impl<'s, 'r, R: Read> PlainEnumDeserializer<'s, 'r, R> {
+ pub fn new(reader: &'r mut R, schema: &'s EnumSchema) -> Self {
+ Self {
+ symbols: &schema.symbols,
+ reader,
+ }
+ }
+}
+
+impl<'de, 's, 'r, R: Read> EnumAccess<'de> for PlainEnumDeserializer<'s, 'r, R> {
+ type Error = Error;
+ type Variant = Self;
+
+ fn variant_seed<V>(self, seed: V) -> Result<(V::Value, Self::Variant), Self::Error>
+ where
+ V: DeserializeSeed<'de>,
+ {
+ let index = zag_i32(self.reader)?;
+ let index = usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))?;
+ let symbol = self.symbols.get(index).ok_or(Details::EnumSymbolIndex {
+ index,
+ num_variants: self.symbols.len(),
+ })?;
+ Ok((
+ seed.deserialize(IdentifierDeserializer::string(symbol))?,
+ self,
+ ))
+ }
+}
+
+impl<'de, 's, 'r, R: Read> VariantAccess<'de> for PlainEnumDeserializer<'s, 'r, R> {
+ type Error = Error;
+
+ fn unit_variant(self) -> Result<(), Self::Error> {
+ Ok(())
+ }
+
+ fn newtype_variant_seed<T>(self, _: T) -> Result<T::Value, Self::Error>
+ where
+ T: DeserializeSeed<'de>,
+ {
+ let unexp = Unexpected::UnitVariant;
+ Err(serde::de::Error::invalid_type(unexp, &"newtype variant"))
+ }
+
+ fn tuple_variant<V>(self, _: usize, _: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ let unexp = Unexpected::UnitVariant;
+ Err(serde::de::Error::invalid_type(unexp, &"tuple variant"))
+ }
+
+ fn struct_variant<V>(self, _: &'static [&'static str], _: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ let unexp = Unexpected::UnitVariant;
+ Err(serde::de::Error::invalid_type(unexp, &"struct variant"))
+ }
+}
+
+pub struct UnionEnumDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
+ reader: &'r mut R,
+ variants: &'s [Schema],
+ config: Config<'s, S>,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> UnionEnumDeserializer<'s, 'r, R, S> {
+ pub fn new(reader: &'r mut R, schema: &'s UnionSchema, config: Config<'s, S>) -> Self {
+ Self {
+ reader,
+ variants: schema.variants(),
+ config,
+ }
+ }
+}
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> EnumAccess<'de>
+ for UnionEnumDeserializer<'s, 'r, R, S>
+{
+ type Error = Error;
+ type Variant = UnionVariantAccess<'s, 'r, R, S>;
+
+ fn variant_seed<V>(self, seed: V) -> Result<(V::Value, Self::Variant), Self::Error>
+ where
+ V: DeserializeSeed<'de>,
+ {
+ let index = zag_i32(self.reader)?;
+ let index = usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))?;
+ let schema = self.variants.get(index).ok_or(Details::GetUnionVariant {
+ index: index as i64,
+ num_variants: self.variants.len(),
+ })?;
+
+ Ok((
+ seed.deserialize(IdentifierDeserializer::index(index as u32))?,
+ UnionVariantAccess::new(schema, self.reader, self.config)?,
+ ))
+ }
+}
+
+pub struct UnionVariantAccess<'s, 'r, R: Read, S: Borrow<Schema>> {
+ schema: &'s Schema,
+ reader: &'r mut R,
+ config: Config<'s, S>,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> UnionVariantAccess<'s, 'r, R, S> {
+ pub fn new(
+ schema: &'s Schema,
+ reader: &'r mut R,
+ config: Config<'s, S>,
+ ) -> Result<Self, Error> {
+ let schema = if let Schema::Ref { name } = schema {
+ config.get_schema(name)?
+ } else {
+ schema
+ };
+ Ok(Self {
+ schema,
+ reader,
+ config,
+ })
+ }
+
+ fn error(&self, ty: &'static str, error: impl Into<String>) -> Error {
+ Error::new(Details::DeserializeSchemaAware {
+ value_type: ty,
+ value: error.into(),
+ schema: self.schema.clone(),
+ })
+ }
+}
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> VariantAccess<'de>
+ for UnionVariantAccess<'s, 'r, R, S>
+{
+ type Error = Error;
+
+ fn unit_variant(self) -> Result<(), Self::Error> {
+ match self.schema {
+ Schema::Null => Ok(()),
+ Schema::Record(record) if record.fields.is_empty() => Ok(()),
+ _ => Err(self.error(
+ "unit variant",
+ "Expected Schema::Null | Schema::Record(fields.len() == 0)",
+ )),
+ }
+ }
+
+ fn newtype_variant_seed<T>(self, seed: T) -> Result<T::Value, Self::Error>
+ where
+ T: DeserializeSeed<'de>,
+ {
+ match self.schema {
+ Schema::Record(record)
+ if record.fields.len() == 1
+ && record
+ .attributes
+ .get("org.apache.avro.rust.union_of_records")
+ == Some(&serde_json::Value::Bool(true)) =>
+ {
+ seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ &record.fields[0].schema,
+ self.config,
+ )?)
+ }
+ _ => seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ self.schema,
+ self.config,
+ )?),
+ }
+ }
+
+ fn tuple_variant<V>(self, len: usize, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ SchemaAwareDeserializer::new(self.reader, self.schema, self.config)?
+ .deserialize_tuple(len, visitor)
+ }
+
+ fn struct_variant<V>(
+ self,
+ fields: &'static [&'static str],
+ visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ SchemaAwareDeserializer::new(self.reader, self.schema, self.config)?.deserialize_struct(
+ DESERIALIZE_ANY,
+ fields,
+ visitor,
+ )
+ }
+}
diff --git a/avro/src/serde/deser_schema/identifier.rs b/avro/src/serde/deser_schema/identifier.rs
new file mode 100644
index 0000000..de51f1a
--- /dev/null
+++ b/avro/src/serde/deser_schema/identifier.rs
@@ -0,0 +1,269 @@
+// 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 crate::Error;
+use crate::error::Details;
+use serde::Deserializer;
+use serde::de::Visitor;
+
+/// Deserializer that only accepts `deserialize_identifier` calls.
+pub enum IdentifierDeserializer<'s> {
+ String(&'s str),
+ Index(u32),
+}
+
+impl<'s> IdentifierDeserializer<'s> {
+ pub fn string(name: &'s str) -> Self {
+ Self::String(name)
+ }
+
+ pub fn index(index: u32) -> Self {
+ Self::Index(index)
+ }
+
+ fn error(&self, error: &'static str) -> Error {
+ Error::new(Details::DeserializeIdentifier(error))
+ }
+}
+
+impl<'de, 's> Deserializer<'de> for IdentifierDeserializer<'s> {
+ type Error = Error;
+
+ fn deserialize_any<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ self.deserialize_identifier(visitor)
+ }
+
+ fn deserialize_bool<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_bool"))
+ }
+
+ fn deserialize_i8<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_i8"))
+ }
+
+ fn deserialize_i16<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_i16"))
+ }
+ fn deserialize_i32<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_i32"))
+ }
+
+ fn deserialize_i64<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_i64"))
+ }
+
+ fn deserialize_u8<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_u8"))
+ }
+
+ fn deserialize_u16<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_u16"))
+ }
+ fn deserialize_u32<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_u32"))
+ }
+
+ fn deserialize_u64<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_u64"))
+ }
+
+ fn deserialize_f32<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_f32"))
+ }
+
+ fn deserialize_f64<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_f64"))
+ }
+
+ fn deserialize_char<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_char"))
+ }
+
+ fn deserialize_str<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_str"))
+ }
+
+ fn deserialize_string<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_string"))
+ }
+
+ fn deserialize_bytes<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_bytes"))
+ }
+
+ fn deserialize_byte_buf<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_byte_buf"))
+ }
+ fn deserialize_option<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_option"))
+ }
+
+ fn deserialize_unit<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_unit"))
+ }
+
+ fn deserialize_unit_struct<V>(
+ self,
+ _name: &'static str,
+ _visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_unit_struct"))
+ }
+ fn deserialize_newtype_struct<V>(
+ self,
+ _name: &'static str,
+ _visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_newtype_struct"))
+ }
+
+ fn deserialize_seq<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_seq"))
+ }
+
+ fn deserialize_tuple<V>(self, _len: usize, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_tuple"))
+ }
+
+ fn deserialize_tuple_struct<V>(
+ self,
+ _name: &'static str,
+ _len: usize,
+ _visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_tuple_struct"))
+ }
+
+ fn deserialize_map<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_map"))
+ }
+
+ fn deserialize_struct<V>(
+ self,
+ _name: &'static str,
+ _fields: &'static [&'static str],
+ _visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_struct"))
+ }
+ fn deserialize_enum<V>(
+ self,
+ _name: &'static str,
+ _variants: &'static [&'static str],
+ _visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_enum"))
+ }
+
+ fn deserialize_identifier<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self {
+ IdentifierDeserializer::String(string) => visitor.visit_str(string),
+ IdentifierDeserializer::Index(index) => visitor.visit_u32(index),
+ }
+ }
+
+ fn deserialize_ignored_any<V>(self, _visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ Err(self.error("deserialize_ignored_any"))
+ }
+}
diff --git a/avro/src/serde/deser_schema/map.rs b/avro/src/serde/deser_schema/map.rs
new file mode 100644
index 0000000..90508da
--- /dev/null
+++ b/avro/src/serde/deser_schema/map.rs
@@ -0,0 +1,147 @@
+// 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 crate::schema::MapSchema;
+use crate::serde::deser_schema::{Config, SchemaAwareDeserializer};
+use crate::util::{zag_i32, zag_i64};
+use crate::{Error, Schema};
+use serde::de::{DeserializeSeed, MapAccess};
+use std::borrow::Borrow;
+use std::io::Read;
+
+pub struct MapDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
+ reader: &'r mut R,
+ schema: &'s Schema,
+ config: Config<'s, S>,
+ remaining: Option<u32>,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> MapDeserializer<'s, 'r, R, S> {
+ pub fn new(
+ reader: &'r mut R,
+ schema: &'s MapSchema,
+ config: Config<'s, S>,
+ ) -> Result<Self, Error> {
+ let schema = if let Schema::Ref { name } = schema.types.as_ref() {
+ config.get_schema(name)?
+ } else {
+ &schema.types
+ };
+ let remaining = Self::read_block_header(reader)?;
+ Ok(Self {
+ schema,
+ reader,
+ config,
+ remaining,
+ })
+ }
+
+ fn read_block_header(reader: &mut R) -> Result<Option<u32>, Error> {
+ let remaining = zag_i32(reader)?;
+ if remaining < 0 {
+ // If the block size is negative the next number is the size of the block in bytes
+ let _bytes = zag_i64(reader)?;
+ }
+ if remaining == 0 {
+ // If the block size is zero the array is finished
+ Ok(None)
+ } else {
+ Ok(Some(remaining.unsigned_abs()))
+ }
+ }
+}
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> MapAccess<'de> for MapDeserializer<'s, 'r, R, S> {
+ type Error = Error;
+
+ fn next_key_seed<K>(&mut self, seed: K) -> Result<Option<K::Value>, Self::Error>
+ where
+ K: DeserializeSeed<'de>,
+ {
+ if self.remaining.is_some() {
+ seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ &Schema::String,
+ self.config,
+ )?)
+ .map(Some)
+ } else {
+ Ok(None)
+ }
+ }
+
+ fn next_value_seed<V>(&mut self, seed: V) -> Result<V::Value, Self::Error>
+ where
+ V: DeserializeSeed<'de>,
+ {
+ let mut remaining = self
+ .remaining
+ .expect("next_key returned None, next_value should not have been called");
+ let value = seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ self.schema,
+ self.config,
+ )?)?;
+
+ remaining -= 1;
+ if remaining == 0 {
+ self.remaining = Self::read_block_header(self.reader)?;
+ } else {
+ self.remaining = Some(remaining);
+ }
+ Ok(value)
+ }
+
+ fn next_entry_seed<K, V>(
+ &mut self,
+ kseed: K,
+ vseed: V,
+ ) -> Result<Option<(K::Value, V::Value)>, Self::Error>
+ where
+ K: DeserializeSeed<'de>,
+ V: DeserializeSeed<'de>,
+ {
+ if let Some(mut remaining) = self.remaining {
+ let key = kseed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ &Schema::String,
+ self.config,
+ )?)?;
+ let value = vseed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ self.schema,
+ self.config,
+ )?)?;
+
+ remaining -= 1;
+ if remaining == 0 {
+ self.remaining = Self::read_block_header(self.reader)?;
+ } else {
+ self.remaining = Some(remaining);
+ }
+
+ Ok(Some((key, value)))
+ } else {
+ Ok(None)
+ }
+ }
+
+ fn size_hint(&self) -> Option<usize> {
+ self.remaining
+ .map(|x| usize::try_from(x).unwrap_or(usize::MAX))
+ }
+}
diff --git a/avro/src/serde/deser_schema/mod.rs b/avro/src/serde/deser_schema/mod.rs
new file mode 100644
index 0000000..99d5897
--- /dev/null
+++ b/avro/src/serde/deser_schema/mod.rs
@@ -0,0 +1,741 @@
+// 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 crate::decode::decode_len;
+use crate::error::Details;
+use crate::schema::{DecimalSchema, InnerDecimalSchema, Name, UnionSchema, UuidSchema};
+use crate::util::{zag_i32, zag_i64};
+use crate::{Error, Schema};
+use serde::de::{Deserializer, Visitor};
+use std::borrow::Borrow;
+use std::collections::HashMap;
+use std::io::Read;
+
+mod array;
+mod enums;
+mod identifier;
+mod map;
+mod record;
+mod tuple;
+
+use crate::serde::deser_schema::enums::UnionEnumDeserializer;
+use array::ArrayDeserializer;
+use enums::PlainEnumDeserializer;
+use map::MapDeserializer;
+use record::RecordDeserializer;
+use tuple::{ManyTupleDeserializer, OneTupleDeserializer};
+
+/// Configure the deserializer.
+#[derive(Debug)]
+pub struct Config<'s, S: Borrow<Schema>> {
+ /// Any references in the schema will be resolved using this map.
+ ///
+ /// This map is not allowed to contain any [`Schema::Ref`], the deserializer is allowed to panic
+ /// in that case.
+ pub names: &'s HashMap<Name, S>,
+ /// Was the data serialized with `human_readable`.
+ pub human_readable: bool,
+}
+
+impl<'s, S: Borrow<Schema>> Config<'s, S> {
+ /// Get the schema for this name.
+ fn get_schema(&self, name: &Name) -> Result<&'s Schema, Error> {
+ self.names
+ .get(name)
+ .map(Borrow::borrow)
+ .ok_or_else(|| Details::SchemaResolutionError(name.clone()).into())
+ }
+}
+
+// This needs to be implemented manually as the derive puts a bound on `S`
+// which is unnecessary as a reference is always Copy.
+impl<'s, S: Borrow<Schema>> Copy for Config<'s, S> {}
+impl<'s, S: Borrow<Schema>> Clone for Config<'s, S> {
+ fn clone(&self) -> Self {
+ *self
+ }
+}
+
+/// A deserializer that deserializes directly from raw Avro datum.
+pub struct SchemaAwareDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
+ reader: &'r mut R,
+ /// The schema of the data being deserialized.
+ ///
+ /// This schema is guaranteed to not be a [`Schema::Ref`].
+ schema: &'s Schema,
+ config: Config<'s, S>,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> SchemaAwareDeserializer<'s, 'r, R, S> {
+ /// Create a new deserializer for this schema.
+ ///
+ /// This will resolve a [`Schema::Ref`] to its actual schema.
+ pub fn new(
+ reader: &'r mut R,
+ schema: &'s Schema,
+ config: Config<'s, S>,
+ ) -> Result<Self, Error> {
+ if let Schema::Ref { name } = schema {
+ let schema = config.get_schema(name)?;
+ Ok(Self {
+ reader,
+ schema,
+ config,
+ })
+ } else {
+ Ok(Self {
+ reader,
+ schema,
+ config,
+ })
+ }
+ }
+
+ /// Create an error for the current type being deserialized with the given message.
+ ///
+ /// This will also include the current schema.
+ fn error(&self, ty: &'static str, error: impl Into<String>) -> Error {
+ Error::new(Details::DeserializeSchemaAware {
+ value_type: ty,
+ value: error.into(),
+ schema: self.schema.clone(),
+ })
+ }
+
+ /// Create a new deserializer with the existing reader and config.
+ ///
+ /// This will resolve the schema if it is a reference.
+ fn with_different_schema(mut self, schema: &'s Schema) -> Result<Self, Error> {
+ self.schema = if let Schema::Ref { name } = schema {
+ self.config.get_schema(name)?
+ } else {
+ schema
+ };
+ Ok(self)
+ }
+
+ /// Read the union and create a new deserializer with the existing reader and config.
+ ///
+ /// This will resolve the read schema if it is a reference.
+ fn with_union(self, schema: &'s UnionSchema) -> Result<Self, Error> {
+ let index = zag_i32(self.reader)?;
+ let variant = schema.get_variant(index as usize)?;
+ self.with_different_schema(variant)
+ }
+
+ /// Read an integer from the reader.
+ ///
+ /// This will check that the current schema is [`Schema::Int`] or a logical type based on that.
+ /// It does not read [`Schema::Union`]s.
+ fn checked_read_int(&mut self, original_ty: &'static str) -> Result<i32, Error> {
+ match self.schema {
+ Schema::Int | Schema::Date | Schema::TimeMillis => zag_i32(self.reader),
+ _ => Err(self.error(
+ original_ty,
+ "Expected Schema::Int | Schema::Date | Schema::TimeMillis",
+ )),
+ }
+ }
+
+ /// Read a long from the reader.
+ ///
+ /// This will check that the current schema is [`Schema::Long`] or a logical type based on that.
+ /// It does not read [`Schema::Union`]s.
+ fn checked_read_long(&mut self, original_ty: &'static str) -> Result<i64, Error> {
+ match self.schema {
+ Schema::Long | Schema::TimeMicros | Schema::TimestampMillis | Schema::TimestampMicros
+ | Schema::TimestampNanos | Schema::LocalTimestampMillis | Schema::LocalTimestampMicros
+ | Schema::LocalTimestampNanos => zag_i64(self.reader),
+ _ => Err(self.error(
+ original_ty,
+ "Expected Schema::Long | Schema::TimeMicros | Schema::{,Local}Timestamp{Millis,Micros,Nanos}",
+ )),
+
+ }
+ }
+
+ /// Read a string from the reader.
+ ///
+ /// This does not check the current schema.
+ fn read_string(&mut self) -> Result<String, Error> {
+ let bytes = self.read_bytes_with_len()?;
+ Ok(String::from_utf8(bytes).map_err(Details::ConvertToUtf8)?)
+ }
+
+ /// Read a bytes from the reader.
+ ///
+ /// This does not check the current schema.
+ fn read_bytes_with_len(&mut self) -> Result<Vec<u8>, Error> {
+ let length = decode_len(self.reader)?;
+ self.read_bytes(length)
+ }
+
+ /// Read `n` bytes from the reader.
+ ///
+ /// This does not check the current schema.
+ fn read_bytes(&mut self, length: usize) -> Result<Vec<u8>, Error> {
+ let mut buf = vec![0; length];
+ self.reader
+ .read_exact(&mut buf)
+ .map_err(Details::ReadBytes)?;
+ Ok(buf)
+ }
+
+ /// Read `n` bytes from the reader.
+ ///
+ /// This does not check the current schema.
+ fn read_array<const N: usize>(&mut self) -> Result<[u8; N], Error> {
+ let mut buf = [0; N];
+ self.reader
+ .read_exact(&mut buf)
+ .map_err(Details::ReadBytes)?;
+ Ok(buf)
+ }
+}
+
+/// A static string that will bypass name checks in `deserialize_*` functions.
+///
+/// This is used so that `deserialize_any` can use the `deserialize_*` implementation which take
+/// a static string.
+///
+/// We don't want users to abuse this feature so this value is compared by pointer address, therefore
+/// a user providing the string below will not be able to skip name validation.
+static DESERIALIZE_ANY: &str = "This value is compared by pointer value";
+/// A static array so that `deserialize_any` can call `deserialize_*` functions.
+static DESERIALIZE_ANY_FIELDS: &[&str] = &[];
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> Deserializer<'de>
+ for SchemaAwareDeserializer<'s, 'r, R, S>
+{
+ type Error = Error;
+
+ fn deserialize_any<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Null => self.deserialize_unit(visitor),
+ Schema::Boolean => self.deserialize_bool(visitor),
+ Schema::Int | Schema::Date | Schema::TimeMillis => self.deserialize_i32(visitor),
+ Schema::Long
+ | Schema::TimeMicros
+ | Schema::TimestampMillis
+ | Schema::TimestampMicros
+ | Schema::TimestampNanos
+ | Schema::LocalTimestampMillis
+ | Schema::LocalTimestampMicros
+ | Schema::LocalTimestampNanos => self.deserialize_i64(visitor),
+ Schema::Float => self.deserialize_f32(visitor),
+ Schema::Double => self.deserialize_f64(visitor),
+ Schema::Bytes
+ | Schema::Fixed(_)
+ | Schema::Decimal(_)
+ | Schema::BigDecimal
+ | Schema::Uuid(UuidSchema::Fixed(_) | UuidSchema::Bytes)
+ | Schema::Duration(_) => self.deserialize_byte_buf(visitor),
+ Schema::String | Schema::Uuid(UuidSchema::String) => self.deserialize_string(visitor),
+ Schema::Array(_) => self.deserialize_seq(visitor),
+ Schema::Map(_) => self.deserialize_map(visitor),
+ Schema::Union(union) => self.with_union(union)?.deserialize_any(visitor),
+ Schema::Record(schema) => {
+ if schema.attributes.get("org.apache.avro.rust.tuple")
+ == Some(&serde_json::Value::Bool(true))
+ {
+ // This attribute is needed because we can't tell the difference between a tuple
+ // and struct, but a tuple needs to be deserialized as a sequence instead of a map.
+ self.deserialize_tuple(schema.fields.len(), visitor)
+ } else {
+ self.deserialize_struct(DESERIALIZE_ANY, DESERIALIZE_ANY_FIELDS, visitor)
+ }
+ }
+ Schema::Enum(_) => {
+ self.deserialize_enum(DESERIALIZE_ANY, DESERIALIZE_ANY_FIELDS, visitor)
+ }
+ Schema::Ref { .. } => unreachable!("References are resolved on deserializer creation"),
+ }
+ }
+
+ fn deserialize_bool<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Boolean => {
+ let mut buf = [0xFF];
+ self.reader
+ .read_exact(&mut buf)
+ .map_err(Details::ReadBytes)?;
+ match buf[0] {
+ 0 => visitor.visit_bool(false),
+ 1 => visitor.visit_bool(true),
+ _ => Err(self.error("bool", format!("{} is not a valid boolean", buf[0]))),
+ }
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_bool(visitor),
+ _ => Err(self.error("bool", "Expected Schema::Boolean")),
+ }
+ }
+
+ fn deserialize_i8<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema {
+ self.with_union(union)?.deserialize_i8(visitor)
+ } else {
+ let int = self.checked_read_int("i8")?;
+ let value = i8::try_from(int)
+ .map_err(|_| self.error("i8", format!("Could not convert int ({int}) to an i8")))?;
+ visitor.visit_i8(value)
+ }
+ }
+
+ fn deserialize_i16<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema {
+ self.with_union(union)?.deserialize_i16(visitor)
+ } else {
+ let int = self.checked_read_int("i16")?;
+ let value = i16::try_from(int).map_err(|_| {
+ self.error("i16", format!("Could not convert int ({int}) to an i16"))
+ })?;
+ visitor.visit_i16(value)
+ }
+ }
+
+ fn deserialize_i32<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema {
+ self.with_union(union)?.deserialize_i32(visitor)
+ } else {
+ visitor.visit_i32(self.checked_read_int("i32")?)
+ }
+ }
+
+ fn deserialize_i64<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema {
+ self.with_union(union)?.deserialize_i64(visitor)
+ } else {
+ visitor.visit_i64(self.checked_read_long("i64")?)
+ }
+ }
+
+ fn deserialize_i128<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Fixed(fixed) if fixed.size == 16 && fixed.name.name() == "i128" => {
+ visitor.visit_i128(i128::from_le_bytes(self.read_array()?))
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_i128(visitor),
+ _ => Err(self.error("i128", r#"Expected Schema::Fixed(name: "i128", size: 16)"#)),
+ }
+ }
+
+ fn deserialize_u8<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema {
+ self.with_union(union)?.deserialize_u8(visitor)
+ } else {
+ let int = self.checked_read_int("u8")?;
+ let value = u8::try_from(int)
+ .map_err(|_| self.error("u8", format!("Could not convert int ({int}) to an u8")))?;
+ visitor.visit_u8(value)
+ }
+ }
+
+ fn deserialize_u16<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema {
+ self.with_union(union)?.deserialize_u16(visitor)
+ } else {
+ let int = self.checked_read_int("u16")?;
+ let value = u16::try_from(int).map_err(|_| {
+ self.error("u16", format!("Could not convert int ({int}) to an u16"))
+ })?;
+ visitor.visit_u16(value)
+ }
+ }
+
+ fn deserialize_u32<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema {
+ self.with_union(union)?.deserialize_u32(visitor)
+ } else {
+ let long = self.checked_read_long("u32")?;
+ let value = u32::try_from(long).map_err(|_| {
+ self.error("u32", format!("Could not convert long ({long}) to an u32"))
+ })?;
+ visitor.visit_u32(value)
+ }
+ }
+
+ fn deserialize_u64<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Fixed(fixed) if fixed.size == 8 && fixed.name.name() == "u64" => {
+ visitor.visit_u64(u64::from_le_bytes(self.read_array()?))
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_u64(visitor),
+ _ => Err(self.error("u64", r#"Expected Schema::Fixed(name: "u64", size: 8)"#)),
+ }
+ }
+
+ fn deserialize_u128<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Fixed(fixed) if fixed.size == 16 && fixed.name.name() == "u128" => {
+ visitor.visit_u128(u128::from_le_bytes(self.read_array()?))
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_i128(visitor),
+ _ => Err(self.error("u128", r#"Expected Schema::Fixed(name: "u128", size: 16)"#)),
+ }
+ }
+
+ fn deserialize_f32<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Float => visitor.visit_f32(f32::from_le_bytes(self.read_array()?)),
+ Schema::Union(union) => self.with_union(union)?.deserialize_f32(visitor),
+ _ => Err(self.error("f32", "Expected Schema::Float")),
+ }
+ }
+
+ fn deserialize_f64<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Double => visitor.visit_f64(f64::from_le_bytes(self.read_array()?)),
+ Schema::Union(union) => self.with_union(union)?.deserialize_f64(visitor),
+ _ => Err(self.error("f64", "Expected Schema::Double")),
+ }
+ }
+
+ fn deserialize_char<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ // A char cannot be deserialized using Schema::Uuid(UuidSchema::String) as that is at least
+ // 32 characters.
+ Schema::String => {
+ let string = self.read_string()?;
+ let mut chars = string.chars();
+ let char = chars
+ .next()
+ .ok_or_else(|| self.error("char", "String is empty"))?;
+ // We can't just check the string length, as that is in bytes not characters
+ if chars.next().is_some() {
+ Err(self.error(
+ "char",
+ format!(r#"Read more than one character: "{string}""#),
+ ))
+ } else {
+ visitor.visit_char(char)
+ }
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_char(visitor),
+ _ => Err(self.error("char", "Expected Schema::String")),
+ }
+ }
+
+ fn deserialize_str<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ self.deserialize_string(visitor)
+ }
+
+ fn deserialize_string<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::String | Schema::Uuid(UuidSchema::String) => {
+ visitor.visit_string(self.read_string()?)
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_string(visitor),
+ _ => Err(self.error("string", "Expected Schema::String | Schema::Uuid(String)")),
+ }
+ }
+
+ fn deserialize_bytes<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ self.deserialize_byte_buf(visitor)
+ }
+
+ fn deserialize_byte_buf<V>(mut self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Bytes | Schema::BigDecimal | Schema::Decimal(DecimalSchema { inner: InnerDecimalSchema::Bytes, ..}) | Schema::Uuid(UuidSchema::Bytes) => {
+ visitor.visit_byte_buf(self.read_bytes_with_len()?)
+ }
+ Schema::Fixed(fixed) | Schema::Decimal(DecimalSchema { inner: InnerDecimalSchema::Fixed(fixed), ..}) | Schema::Uuid(UuidSchema::Fixed(fixed)) | Schema::Duration(fixed) => {
+ visitor.visit_byte_buf(self.read_bytes(fixed.size)?)
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_byte_buf(visitor),
+ _ => Err(self.error("bytes", "Expected Schema::Bytes | Schema::Fixed | Schema::BigDecimal | Schema::Decimal | Schema::Uuid(Fixed | Bytes) | Schema::Duration")),
+ }
+ }
+
+ fn deserialize_option<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ if let Schema::Union(union) = self.schema
+ && union.variants().len() == 2
+ && union.is_nullable()
+ {
+ let index = zag_i32(self.reader)?;
+ let schema = union.get_variant(index as usize)?;
+ if let Schema::Null = schema {
+ visitor.visit_none()
+ } else {
+ visitor.visit_some(self.with_different_schema(schema)?)
+ }
+ } else {
+ Err(self.error("option", "Expected Schema::Union([Schema::Null, _])"))
+ }
+ }
+
+ fn deserialize_unit<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Null => visitor.visit_unit(),
+ Schema::Union(union) => self.with_union(union)?.deserialize_unit(visitor),
+ _ => Err(self.error("unit", "Expected Schema::Null")),
+ }
+ }
+
+ fn deserialize_unit_struct<V>(
+ self,
+ name: &'static str,
+ visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Record(record) if record.fields.is_empty() && record.name.name() == name => {
+ visitor.visit_unit()
+ }
+ Schema::Union(union) => self
+ .with_union(union)?
+ .deserialize_unit_struct(name, visitor),
+ _ => Err(self.error(
+ "unit struct",
+ format!("Expected Schema::Record(name: {name}, fields.len() == 0)"),
+ )),
+ }
+ }
+
+ fn deserialize_newtype_struct<V>(
+ self,
+ name: &'static str,
+ visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Record(record) if record.fields.len() == 1 && record.name.name() == name => {
+ visitor.visit_newtype_struct(self.with_different_schema(&record.fields[0].schema)?)
+ }
+ Schema::Union(union) => self
+ .with_union(union)?
+ .deserialize_newtype_struct(name, visitor),
+ _ => Err(self.error(
+ "newtype struct",
+ format!("Expected Schema::Record(name: {name}, fields.len() == 1)"),
+ )),
+ }
+ }
+
+ fn deserialize_seq<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Array(array) => {
+ visitor.visit_seq(ArrayDeserializer::new(self.reader, array, self.config)?)
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_seq(visitor),
+ _ => Err(self.error("seq", "Expected Schema::Array")),
+ }
+ }
+
+ fn deserialize_tuple<V>(self, len: usize, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ // `len == 0` is not possible for derived Deserialize implementations but users might use it.
+ // The derived Deserialize implementations use `deserialize_unit` instead
+ Schema::Null if len == 0 => visitor.visit_unit(),
+ schema if len == 1 => {
+ visitor.visit_seq(OneTupleDeserializer::new(self.reader, schema, self.config)?)
+ }
+ Schema::Record(record) if record.fields.len() == len => {
+ visitor.visit_seq(ManyTupleDeserializer::new(self.reader, record, self.config))
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_tuple(len, visitor),
+ _ if len == 0 => Err(self.error("tuple", "Expected Schema::Null for unit tuple")),
+ _ => Err(self.error(
+ "tuple",
+ format!("Expected Schema::Record(fields.len() == {len}) for {len}-tuple"),
+ )),
+ }
+ }
+
+ fn deserialize_tuple_struct<V>(
+ self,
+ name: &'static str,
+ len: usize,
+ visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Record(record) if record.name.name() == name && record.fields.len() == len => {
+ visitor.visit_map(RecordDeserializer::new(self.reader, record, self.config))
+ }
+ Schema::Union(union) => self
+ .with_union(union)?
+ .deserialize_tuple_struct(name, len, visitor),
+ _ => Err(self.error(
+ "tuple struct",
+ format!("Expected Schema::Record(name: {name}, fields.len() == {len})"),
+ )),
+ }
+ }
+
+ fn deserialize_map<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Map(map) => {
+ visitor.visit_map(MapDeserializer::new(self.reader, map, self.config)?)
+ }
+ Schema::Record(record) => {
+ // Needed for flattened structs which are (de)serialized as maps
+ visitor.visit_map(RecordDeserializer::new(self.reader, record, self.config))
+ }
+ Schema::Union(union) => self.with_union(union)?.deserialize_map(visitor),
+ _ => Err(self.error("map", "Expected Schema::Map")),
+ }
+ }
+
+ fn deserialize_struct<V>(
+ self,
+ name: &'static str,
+ fields: &'static [&'static str],
+ visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ // Checking that the amount of fields match does not work because of `skip_deserializing`
+ Schema::Record(record)
+ if record.name.name() == name || name.as_ptr() == DESERIALIZE_ANY.as_ptr() =>
+ {
+ visitor.visit_map(RecordDeserializer::new(self.reader, record, self.config))
+ }
+ Schema::Union(union) => self
+ .with_union(union)?
+ .deserialize_struct(name, fields, visitor),
+ _ => Err(self.error("struct", format!("Expected Schema::Record(name: {name})"))),
+ }
+ }
+
+ fn deserialize_enum<V>(
+ self,
+ _name: &'static str,
+ variants: &'static [&'static str],
+ visitor: V,
+ ) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::Enum(schema) if schema.symbols.len() == variants.len() => {
+ visitor.visit_enum(PlainEnumDeserializer::new(self.reader, schema))
+ }
+ Schema::Union(union) if union.variants().len() == variants.len() => {
+ visitor.visit_enum(UnionEnumDeserializer::new(self.reader, union, self.config))
+ }
+ _ => Err(self.error(
+ "enum",
+ format!(
+ "Expected Schema::Enum(symbols.len() == {}) | Schema::Union(variants.len() == {}",
+ variants.len(), variants.len()
+ ),
+ )),
+ }
+ }
+
+ fn deserialize_identifier<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ match self.schema {
+ Schema::String => self.deserialize_string(visitor),
+ _ => Err(self.error("identifier", "Expected Schema::String")),
+ }
+ }
+
+ fn deserialize_ignored_any<V>(self, visitor: V) -> Result<V::Value, Self::Error>
+ where
+ V: Visitor<'de>,
+ {
+ // TODO: Add `Seek` bound and skip over ignored data
+ self.deserialize_any(visitor)
+ }
+
+ fn is_human_readable(&self) -> bool {
+ self.config.human_readable
+ }
+}
diff --git a/avro/src/serde/deser_schema/record.rs b/avro/src/serde/deser_schema/record.rs
new file mode 100644
index 0000000..2414037
--- /dev/null
+++ b/avro/src/serde/deser_schema/record.rs
@@ -0,0 +1,93 @@
+// 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 crate::schema::RecordSchema;
+use crate::serde::deser_schema::identifier::IdentifierDeserializer;
+use crate::serde::deser_schema::{Config, SchemaAwareDeserializer};
+use crate::{Error, Schema};
+use serde::de::{DeserializeSeed, MapAccess};
+use std::borrow::Borrow;
+use std::io::Read;
+
+pub struct RecordDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
+ reader: &'r mut R,
+ schema: &'s RecordSchema,
+ config: Config<'s, S>,
+ current_field: usize,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> RecordDeserializer<'s, 'r, R, S> {
+ pub fn new(reader: &'r mut R, schema: &'s RecordSchema, config: Config<'s, S>) -> Self {
+ Self {
+ reader,
+ schema,
+ config,
+ current_field: 0,
+ }
+ }
+}
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> MapAccess<'de> for RecordDeserializer<'s, 'r, R, S> {
+ type Error = Error;
+
+ fn next_key_seed<K>(&mut self, seed: K) -> Result<Option<K::Value>, Self::Error>
+ where
+ K: DeserializeSeed<'de>,
+ {
+ if self.current_field >= self.schema.fields.len() {
+ // Finished reading this record
+ Ok(None)
+ } else {
+ seed.deserialize(IdentifierDeserializer::string(
+ &self.schema.fields[self.current_field].name,
+ ))
+ .map(Some)
+ }
+ }
+
+ fn next_value_seed<V>(&mut self, seed: V) -> Result<V::Value, Self::Error>
+ where
+ V: DeserializeSeed<'de>,
+ {
+ let schema = &self.schema.fields[self.current_field].schema;
+ let value = if let Schema::Record(record) = schema
+ && record.fields.len() == 1
+ && record
+ .attributes
+ .get("org.apache.avro.rust.union_of_records")
+ == Some(&serde_json::Value::Bool(true))
+ {
+ seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ &record.fields[0].schema,
+ self.config,
+ )?)?
+ } else {
+ seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ schema,
+ self.config,
+ )?)?
+ };
+ self.current_field += 1;
+ Ok(value)
+ }
+
+ fn size_hint(&self) -> Option<usize> {
+ Some(self.schema.fields.len() - self.current_field)
+ }
+}
diff --git a/avro/src/serde/deser_schema/tuple.rs b/avro/src/serde/deser_schema/tuple.rs
new file mode 100644
index 0000000..5b68746
--- /dev/null
+++ b/avro/src/serde/deser_schema/tuple.rs
@@ -0,0 +1,123 @@
+// 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 crate::schema::RecordSchema;
+use crate::serde::deser_schema::{Config, SchemaAwareDeserializer};
+use crate::{Error, Schema};
+use serde::de::{DeserializeSeed, SeqAccess};
+use std::borrow::Borrow;
+use std::io::Read;
+
+pub struct OneTupleDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
+ reader: &'r mut R,
+ schema: &'s Schema,
+ config: Config<'s, S>,
+ field_read: bool,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> OneTupleDeserializer<'s, 'r, R, S> {
+ pub fn new(
+ reader: &'r mut R,
+ schema: &'s Schema,
+ config: Config<'s, S>,
+ ) -> Result<Self, Error> {
+ let schema = if let Schema::Ref { name } = schema {
+ config.get_schema(name)?
+ } else {
+ schema
+ };
+ Ok(Self {
+ reader,
+ schema,
+ config,
+ field_read: false,
+ })
+ }
+}
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> SeqAccess<'de>
+ for OneTupleDeserializer<'s, 'r, R, S>
+{
+ type Error = Error;
+
+ fn next_element_seed<T>(&mut self, seed: T) -> Result<Option<T::Value>, Self::Error>
+ where
+ T: DeserializeSeed<'de>,
+ {
+ if self.field_read {
+ Ok(None)
+ } else {
+ self.field_read = true;
+ seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ self.schema,
+ self.config,
+ )?)
+ .map(Some)
+ }
+ }
+
+ fn size_hint(&self) -> Option<usize> {
+ Some(1 - usize::from(self.field_read))
+ }
+}
+
+pub struct ManyTupleDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
+ reader: &'r mut R,
+ schema: &'s RecordSchema,
+ config: Config<'s, S>,
+ current_field: usize,
+}
+
+impl<'s, 'r, R: Read, S: Borrow<Schema>> ManyTupleDeserializer<'s, 'r, R, S> {
+ pub fn new(reader: &'r mut R, schema: &'s RecordSchema, config: Config<'s, S>) -> Self {
+ Self {
+ reader,
+ schema,
+ config,
+ current_field: 0,
+ }
+ }
+}
+
+impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> SeqAccess<'de>
+ for ManyTupleDeserializer<'s, 'r, R, S>
+{
+ type Error = Error;
+
+ fn next_element_seed<T>(&mut self, seed: T) -> Result<Option<T::Value>, Self::Error>
+ where
+ T: DeserializeSeed<'de>,
+ {
+ if self.current_field < self.schema.fields.len() {
+ let schema = &self.schema.fields[self.current_field].schema;
+ self.current_field += 1;
+ seed.deserialize(SchemaAwareDeserializer::new(
+ self.reader,
+ schema,
+ self.config,
+ )?)
+ .map(Some)
+ } else {
+ Ok(None)
+ }
+ }
+
+ fn size_hint(&self) -> Option<usize> {
+ todo!()
+ }
+}
diff --git a/avro/src/serde/mod.rs b/avro/src/serde/mod.rs
index b3bfd2a..256963d 100644
--- a/avro/src/serde/mod.rs
+++ b/avro/src/serde/mod.rs
@@ -109,6 +109,7 @@
mod de;
mod derive;
+pub(crate) mod deser_schema;
mod ser;
pub(crate) mod ser_schema;
mod util;