// 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.

//! Logic handling the intermediate representation of Avro values.
use crate::{
    decimal::Decimal,
    duration::Duration,
    schema::{
        DecimalSchema, EnumSchema, FixedSchema, Name, Namespace, Precision, RecordField,
        RecordSchema, ResolvedSchema, Scale, Schema, SchemaKind, UnionSchema,
    },
    AvroResult, Error,
};
use serde_json::{Number, Value as JsonValue};
use std::{
    borrow::Borrow,
    collections::{BTreeMap, HashMap},
    convert::TryFrom,
    fmt::Debug,
    hash::BuildHasher,
    str::FromStr,
};
use uuid::Uuid;

/// Compute the maximum decimal value precision of a byte array of length `len` could hold.
fn max_prec_for_len(len: usize) -> Result<usize, Error> {
    let len = i32::try_from(len).map_err(|e| Error::ConvertLengthToI32(e, len))?;
    Ok((2.0_f64.powi(8 * len - 1) - 1.0).log10().floor() as usize)
}

/// A valid Avro value.
///
/// More information about Avro values can be found in the [Avro
/// Specification](https://avro.apache.org/docs/current/spec.html#schemas)
#[derive(Clone, Debug, PartialEq, strum_macros::EnumDiscriminants)]
#[strum_discriminants(name(ValueKind))]
pub enum Value {
    /// A `null` Avro value.
    Null,
    /// A `boolean` Avro value.
    Boolean(bool),
    /// A `int` Avro value.
    Int(i32),
    /// A `long` Avro value.
    Long(i64),
    /// A `float` Avro value.
    Float(f32),
    /// A `double` Avro value.
    Double(f64),
    /// A `bytes` Avro value.
    Bytes(Vec<u8>),
    /// A `string` Avro value.
    String(String),
    /// A `fixed` Avro value.
    /// The size of the fixed value is represented as a `usize`.
    Fixed(usize, Vec<u8>),
    /// An `enum` Avro value.
    ///
    /// An Enum is represented by a symbol and its position in the symbols list
    /// of its corresponding schema.
    /// This allows schema-less encoding, as well as schema resolution while
    /// reading values.
    Enum(u32, String),
    /// An `union` Avro value.
    ///
    /// A Union is represented by the value it holds and its position in the type list
    /// of its corresponding schema
    /// This allows schema-less encoding, as well as schema resolution while
    /// reading values.
    Union(u32, Box<Value>),
    /// An `array` Avro value.
    Array(Vec<Value>),
    /// A `map` Avro value.
    Map(HashMap<String, Value>),
    /// A `record` Avro value.
    ///
    /// A Record is represented by a vector of (`<record name>`, `value`).
    /// This allows schema-less encoding.
    ///
    /// See [Record](types.Record) for a more user-friendly support.
    Record(Vec<(String, Value)>),
    /// A date value.
    ///
    /// Serialized and deserialized as `i32` directly. Can only be deserialized properly with a
    /// schema.
    Date(i32),
    /// An Avro Decimal value. Bytes are in big-endian order, per the Avro spec.
    Decimal(Decimal),
    /// Time in milliseconds.
    TimeMillis(i32),
    /// Time in microseconds.
    TimeMicros(i64),
    /// Timestamp in milliseconds.
    TimestampMillis(i64),
    /// Timestamp in microseconds.
    TimestampMicros(i64),
    /// Local timestamp in milliseconds.
    LocalTimestampMillis(i64),
    /// Local timestamp in microseconds.
    LocalTimestampMicros(i64),
    /// Avro Duration. An amount of time defined by months, days and milliseconds.
    Duration(Duration),
    /// Universally unique identifier.
    Uuid(Uuid),
}
/// Any structure implementing the [ToAvro](trait.ToAvro.html) trait will be usable
/// from a [Writer](../writer/struct.Writer.html).
#[deprecated(
    since = "0.11.0",
    note = "Please use Value::from, Into::into or value.into() instead"
)]
pub trait ToAvro {
    /// Transforms this value into an Avro-compatible [Value](enum.Value.html).
    fn avro(self) -> Value;
}

#[allow(deprecated)]
impl<T: Into<Value>> ToAvro for T {
    fn avro(self) -> Value {
        self.into()
    }
}

macro_rules! to_value(
    ($type:ty, $variant_constructor:expr) => (
        impl From<$type> for Value {
            fn from(value: $type) -> Self {
                $variant_constructor(value)
            }
        }
    );
);

to_value!(bool, Value::Boolean);
to_value!(i32, Value::Int);
to_value!(i64, Value::Long);
to_value!(f32, Value::Float);
to_value!(f64, Value::Double);
to_value!(String, Value::String);
to_value!(Vec<u8>, Value::Bytes);
to_value!(uuid::Uuid, Value::Uuid);
to_value!(Decimal, Value::Decimal);
to_value!(Duration, Value::Duration);

impl From<()> for Value {
    fn from(_: ()) -> Self {
        Self::Null
    }
}

impl From<usize> for Value {
    fn from(value: usize) -> Self {
        i64::try_from(value)
            .expect("cannot convert usize to i64")
            .into()
    }
}

impl From<&str> for Value {
    fn from(value: &str) -> Self {
        Self::String(value.to_owned())
    }
}

impl From<&[u8]> for Value {
    fn from(value: &[u8]) -> Self {
        Self::Bytes(value.to_owned())
    }
}

impl<T> From<Option<T>> for Value
where
    T: Into<Self>,
{
    fn from(value: Option<T>) -> Self {
        // FIXME: this is incorrect in case first type in union is not "none"
        Self::Union(
            value.is_some() as u32,
            Box::new(value.map_or_else(|| Self::Null, Into::into)),
        )
    }
}

impl<K, V, S> From<HashMap<K, V, S>> for Value
where
    K: Into<String>,
    V: Into<Self>,
    S: BuildHasher,
{
    fn from(value: HashMap<K, V, S>) -> Self {
        Self::Map(
            value
                .into_iter()
                .map(|(key, value)| (key.into(), value.into()))
                .collect(),
        )
    }
}

/// Utility interface to build `Value::Record` objects.
#[derive(Debug, Clone)]
pub struct Record<'a> {
    /// List of fields contained in the record.
    /// Ordered according to the fields in the schema given to create this
    /// `Record` object. Any unset field defaults to `Value::Null`.
    pub fields: Vec<(String, Value)>,
    schema_lookup: &'a BTreeMap<String, usize>,
}

impl<'a> Record<'a> {
    /// Create a `Record` given a `Schema`.
    ///
    /// If the `Schema` is not a `Schema::Record` variant, `None` will be returned.
    pub fn new(schema: &Schema) -> Option<Record> {
        match *schema {
            Schema::Record(RecordSchema {
                fields: ref schema_fields,
                lookup: ref schema_lookup,
                ..
            }) => {
                let mut fields = Vec::with_capacity(schema_fields.len());
                for schema_field in schema_fields.iter() {
                    fields.push((schema_field.name.clone(), Value::Null));
                }

                Some(Record {
                    fields,
                    schema_lookup,
                })
            }
            _ => None,
        }
    }

    /// Put a compatible value (implementing the `ToAvro` trait) in the
    /// `Record` for a given `field` name.
    ///
    /// **NOTE** Only ensure that the field name is present in the `Schema` given when creating
    /// this `Record`. Does not perform any schema validation.
    pub fn put<V>(&mut self, field: &str, value: V)
    where
        V: Into<Value>,
    {
        if let Some(&position) = self.schema_lookup.get(field) {
            self.fields[position].1 = value.into()
        }
    }
}

impl<'a> From<Record<'a>> for Value {
    fn from(value: Record<'a>) -> Self {
        Self::Record(value.fields)
    }
}

impl From<JsonValue> for Value {
    fn from(value: JsonValue) -> Self {
        match value {
            JsonValue::Null => Self::Null,
            JsonValue::Bool(b) => b.into(),
            JsonValue::Number(ref n) if n.is_i64() => Value::Long(n.as_i64().unwrap()),
            JsonValue::Number(ref n) if n.is_f64() => Value::Double(n.as_f64().unwrap()),
            JsonValue::Number(n) => Value::Long(n.as_u64().unwrap() as i64), // TODO: Not so great
            JsonValue::String(s) => s.into(),
            JsonValue::Array(items) => Value::Array(items.into_iter().map(Value::from).collect()),
            JsonValue::Object(items) => Value::Map(
                items
                    .into_iter()
                    .map(|(key, value)| (key, value.into()))
                    .collect(),
            ),
        }
    }
}

/// Convert Avro values to Json values
impl TryFrom<Value> for JsonValue {
    type Error = crate::error::Error;
    fn try_from(value: Value) -> AvroResult<Self> {
        match value {
            Value::Null => Ok(Self::Null),
            Value::Boolean(b) => Ok(Self::Bool(b)),
            Value::Int(i) => Ok(Self::Number(i.into())),
            Value::Long(l) => Ok(Self::Number(l.into())),
            Value::Float(f) => Number::from_f64(f.into())
                .map(Self::Number)
                .ok_or_else(|| Error::ConvertF64ToJson(f.into())),
            Value::Double(d) => Number::from_f64(d)
                .map(Self::Number)
                .ok_or(Error::ConvertF64ToJson(d)),
            Value::Bytes(bytes) => Ok(Self::Array(bytes.into_iter().map(|b| b.into()).collect())),
            Value::String(s) => Ok(Self::String(s)),
            Value::Fixed(_size, items) => {
                Ok(Self::Array(items.into_iter().map(|v| v.into()).collect()))
            }
            Value::Enum(_i, s) => Ok(Self::String(s)),
            Value::Union(_i, b) => Self::try_from(*b),
            Value::Array(items) => items
                .into_iter()
                .map(Self::try_from)
                .collect::<Result<Vec<_>, _>>()
                .map(Self::Array),
            Value::Map(items) => items
                .into_iter()
                .map(|(key, value)| Self::try_from(value).map(|v| (key, v)))
                .collect::<Result<Vec<_>, _>>()
                .map(|v| Self::Object(v.into_iter().collect())),
            Value::Record(items) => items
                .into_iter()
                .map(|(key, value)| Self::try_from(value).map(|v| (key, v)))
                .collect::<Result<Vec<_>, _>>()
                .map(|v| Self::Object(v.into_iter().collect())),
            Value::Date(d) => Ok(Self::Number(d.into())),
            Value::Decimal(ref d) => <Vec<u8>>::try_from(d)
                .map(|vec| Self::Array(vec.into_iter().map(|v| v.into()).collect())),
            Value::TimeMillis(t) => Ok(Self::Number(t.into())),
            Value::TimeMicros(t) => Ok(Self::Number(t.into())),
            Value::TimestampMillis(t) => Ok(Self::Number(t.into())),
            Value::TimestampMicros(t) => Ok(Self::Number(t.into())),
            Value::LocalTimestampMillis(t) => Ok(Self::Number(t.into())),
            Value::LocalTimestampMicros(t) => Ok(Self::Number(t.into())),
            Value::Duration(d) => Ok(Self::Array(
                <[u8; 12]>::from(d).iter().map(|&v| v.into()).collect(),
            )),
            Value::Uuid(uuid) => Ok(Self::String(uuid.as_hyphenated().to_string())),
        }
    }
}

impl Value {
    /// Validate the value against the given [Schema](../schema/enum.Schema.html).
    ///
    /// See the [Avro specification](https://avro.apache.org/docs/current/spec.html)
    /// for the full set of rules of schema validation.
    pub fn validate(&self, schema: &Schema) -> bool {
        self.validate_schemata(vec![schema])
    }

    pub fn validate_schemata(&self, schemata: Vec<&Schema>) -> bool {
        let rs = ResolvedSchema::try_from(schemata.clone())
            .expect("Schemata didn't successfully resolve");
        let schemata_len = schemata.len();
        schemata.iter().any(|schema| {
            let enclosing_namespace = schema.namespace();

            match self.validate_internal(schema, rs.get_names(), &enclosing_namespace) {
                Some(reason) => {
                    let log_message = format!(
                        "Invalid value: {:?} for schema: {:?}. Reason: {}",
                        self, schema, reason
                    );
                    if schemata_len == 1 {
                        error!("{}", log_message);
                    } else {
                        debug!("{}", log_message);
                    };
                    false
                }
                None => true,
            }
        })
    }

    fn accumulate(accumulator: Option<String>, other: Option<String>) -> Option<String> {
        match (accumulator, other) {
            (None, None) => None,
            (None, s @ Some(_)) => s,
            (s @ Some(_), None) => s,
            (Some(reason1), Some(reason2)) => Some(format!("{reason1}\n{reason2}")),
        }
    }

    /// Validates the value against the provided schema.
    pub(crate) fn validate_internal<S: std::borrow::Borrow<Schema> + Debug>(
        &self,
        schema: &Schema,
        names: &HashMap<Name, S>,
        enclosing_namespace: &Namespace,
    ) -> Option<String> {
        match (self, schema) {
            (_, Schema::Ref { name }) => {
                let name = name.fully_qualified_name(enclosing_namespace);
                names.get(&name).map_or_else(
                    || {
                        Some(format!(
                            "Unresolved schema reference: '{:?}'. Parsed names: {:?}",
                            name,
                            names.keys()
                        ))
                    },
                    |s| self.validate_internal(s.borrow(), names, &name.namespace),
                )
            }
            (&Value::Null, &Schema::Null) => None,
            (&Value::Boolean(_), &Schema::Boolean) => None,
            (&Value::Int(_), &Schema::Int) => None,
            (&Value::Int(_), &Schema::Date) => None,
            (&Value::Int(_), &Schema::TimeMillis) => None,
            (&Value::Int(_), &Schema::Long) => None,
            (&Value::Long(_), &Schema::Long) => None,
            (&Value::Long(_), &Schema::TimeMicros) => None,
            (&Value::Long(_), &Schema::TimestampMillis) => None,
            (&Value::Long(_), &Schema::TimestampMicros) => None,
            (&Value::Long(_), &Schema::LocalTimestampMillis) => None,
            (&Value::Long(_), &Schema::LocalTimestampMicros) => None,
            (&Value::TimestampMicros(_), &Schema::TimestampMicros) => None,
            (&Value::TimestampMillis(_), &Schema::TimestampMillis) => None,
            (&Value::LocalTimestampMicros(_), &Schema::LocalTimestampMicros) => None,
            (&Value::LocalTimestampMillis(_), &Schema::LocalTimestampMillis) => None,
            (&Value::TimeMicros(_), &Schema::TimeMicros) => None,
            (&Value::TimeMillis(_), &Schema::TimeMillis) => None,
            (&Value::Date(_), &Schema::Date) => None,
            (&Value::Decimal(_), &Schema::Decimal { .. }) => None,
            (&Value::Duration(_), &Schema::Duration) => None,
            (&Value::Uuid(_), &Schema::Uuid) => None,
            (&Value::Float(_), &Schema::Float) => None,
            (&Value::Float(_), &Schema::Double) => None,
            (&Value::Double(_), &Schema::Double) => None,
            (&Value::Bytes(_), &Schema::Bytes) => None,
            (&Value::Bytes(_), &Schema::Decimal { .. }) => None,
            (&Value::String(_), &Schema::String) => None,
            (&Value::String(_), &Schema::Uuid) => None,
            (&Value::Fixed(n, _), &Schema::Fixed(FixedSchema { size, .. })) => {
                if n != size {
                    Some(format!(
                        "The value's size ({n}) is different than the schema's size ({size})"
                    ))
                } else {
                    None
                }
            }
            (Value::Bytes(b), &Schema::Fixed(FixedSchema { size, .. })) => {
                if b.len() != size {
                    Some(format!(
                        "The bytes' length ({}) is different than the schema's size ({})",
                        b.len(),
                        size
                    ))
                } else {
                    None
                }
            }
            (&Value::Fixed(n, _), &Schema::Duration) => {
                if n != 12 {
                    Some(format!(
                        "The value's size ('{n}') must be exactly 12 to be a Duration"
                    ))
                } else {
                    None
                }
            }
            // TODO: check precision against n
            (&Value::Fixed(_n, _), &Schema::Decimal { .. }) => None,
            (Value::String(s), Schema::Enum(EnumSchema { symbols, .. })) => {
                if !symbols.contains(s) {
                    Some(format!("'{s}' is not a member of the possible symbols"))
                } else {
                    None
                }
            }
            (
                &Value::Enum(i, ref s),
                Schema::Enum(EnumSchema {
                    symbols, default, ..
                }),
            ) => symbols
                .get(i as usize)
                .map(|ref symbol| {
                    if symbol != &s {
                        Some(format!("Symbol '{s}' is not at position '{i}'"))
                    } else {
                        None
                    }
                })
                .unwrap_or_else(|| match default {
                    Some(_) => None,
                    None => Some(format!("No symbol at position '{i}'")),
                }),
            // (&Value::Union(None), &Schema::Union(_)) => None,
            (&Value::Union(i, ref value), Schema::Union(inner)) => inner
                .variants()
                .get(i as usize)
                .map(|schema| value.validate_internal(schema, names, enclosing_namespace))
                .unwrap_or_else(|| Some(format!("No schema in the union at position '{i}'"))),
            (v, Schema::Union(inner)) => {
                match inner.find_schema_with_known_schemata(v, Some(names), enclosing_namespace) {
                    Some(_) => None,
                    None => Some("Could not find matching type in union".to_string()),
                }
            }
            (Value::Array(items), Schema::Array(inner)) => items.iter().fold(None, |acc, item| {
                Value::accumulate(
                    acc,
                    item.validate_internal(inner, names, enclosing_namespace),
                )
            }),
            (Value::Map(items), Schema::Map(inner)) => {
                items.iter().fold(None, |acc, (_, value)| {
                    Value::accumulate(
                        acc,
                        value.validate_internal(inner, names, enclosing_namespace),
                    )
                })
            }
            (
                Value::Record(record_fields),
                Schema::Record(RecordSchema {
                    fields,
                    lookup,
                    name,
                    ..
                }),
            ) => {
                let non_nullable_fields_count =
                    fields.iter().filter(|&rf| !rf.is_nullable()).count();

                // If the record contains fewer fields as required fields by the schema, it is invalid.
                if record_fields.len() < non_nullable_fields_count {
                    return Some(format!(
                        "The value's records length ({}) doesn't match the schema ({} non-nullable fields)",
                        record_fields.len(),
                        non_nullable_fields_count
                    ));
                } else if record_fields.len() > fields.len() {
                    return Some(format!(
                        "The value's records length ({}) is greater than the schema's ({} fields)",
                        record_fields.len(),
                        fields.len(),
                    ));
                }

                record_fields
                    .iter()
                    .fold(None, |acc, (field_name, record_field)| {
                        let record_namespace = if name.namespace.is_none() {
                            enclosing_namespace
                        } else {
                            &name.namespace
                        };
                        match lookup.get(field_name) {
                            Some(idx) => {
                                let field = &fields[*idx];
                                Value::accumulate(
                                    acc,
                                    record_field.validate_internal(
                                        &field.schema,
                                        names,
                                        record_namespace,
                                    ),
                                )
                            }
                            None => Value::accumulate(
                                acc,
                                Some(format!("There is no schema field for field '{field_name}'")),
                            ),
                        }
                    })
            }
            (Value::Map(items), Schema::Record(RecordSchema { fields, .. })) => {
                fields.iter().fold(None, |acc, field| {
                    if let Some(item) = items.get(&field.name) {
                        let res = item.validate_internal(&field.schema, names, enclosing_namespace);
                        Value::accumulate(acc, res)
                    } else if !field.is_nullable() {
                        Value::accumulate(
                            acc,
                            Some(format!(
                                "Field with name '{:?}' is not a member of the map items",
                                field.name
                            )),
                        )
                    } else {
                        acc
                    }
                })
            }
            (_v, _s) => Some("Unsupported value-schema combination".to_string()),
        }
    }

    /// Attempt to perform schema resolution on the value, with the given
    /// [Schema](../schema/enum.Schema.html).
    ///
    /// See [Schema Resolution](https://avro.apache.org/docs/current/spec.html#Schema+Resolution)
    /// in the Avro specification for the full set of rules of schema
    /// resolution.
    pub fn resolve(self, schema: &Schema) -> AvroResult<Self> {
        let enclosing_namespace = schema.namespace();
        let rs = ResolvedSchema::try_from(schema)?;
        self.resolve_internal(schema, rs.get_names(), &enclosing_namespace, &None)
    }

    /// Attempt to perform schema resolution on the value, with the given
    /// [Schema](../schema/enum.Schema.html) and set of schemas to use for Refs resolution.
    ///
    /// See [Schema Resolution](https://avro.apache.org/docs/current/spec.html#Schema+Resolution)
    /// in the Avro specification for the full set of rules of schema
    /// resolution.
    pub fn resolve_schemata(self, schema: &Schema, schemata: Vec<&Schema>) -> AvroResult<Self> {
        let enclosing_namespace = schema.namespace();
        let rs = ResolvedSchema::try_from(schemata)?;
        self.resolve_internal(schema, rs.get_names(), &enclosing_namespace, &None)
    }

    pub(crate) fn resolve_internal<S: Borrow<Schema> + Debug>(
        mut self,
        schema: &Schema,
        names: &HashMap<Name, S>,
        enclosing_namespace: &Namespace,
        field_default: &Option<JsonValue>,
    ) -> AvroResult<Self> {
        // Check if this schema is a union, and if the reader schema is not.
        if SchemaKind::from(&self) == SchemaKind::Union
            && SchemaKind::from(schema) != SchemaKind::Union
        {
            // Pull out the Union, and attempt to resolve against it.
            let v = match self {
                Value::Union(_i, b) => *b,
                _ => unreachable!(),
            };
            self = v;
        }

        match *schema {
            Schema::Ref { ref name } => {
                let name = name.fully_qualified_name(enclosing_namespace);

                if let Some(resolved) = names.get(&name) {
                    debug!("Resolved {:?}", name);
                    self.resolve_internal(resolved.borrow(), names, &name.namespace, field_default)
                } else {
                    error!("Failed to resolve schema {:?}", name);
                    Err(Error::SchemaResolutionError(name.clone()))
                }
            }
            Schema::Null => self.resolve_null(),
            Schema::Boolean => self.resolve_boolean(),
            Schema::Int => self.resolve_int(),
            Schema::Long => self.resolve_long(),
            Schema::Float => self.resolve_float(),
            Schema::Double => self.resolve_double(),
            Schema::Bytes => self.resolve_bytes(),
            Schema::String => self.resolve_string(),
            Schema::Fixed(FixedSchema { size, .. }) => self.resolve_fixed(size),
            Schema::Union(ref inner) => {
                self.resolve_union(inner, names, enclosing_namespace, field_default)
            }
            Schema::Enum(EnumSchema {
                ref symbols,
                ref default,
                ..
            }) => self.resolve_enum(symbols, default, field_default),
            Schema::Array(ref inner) => self.resolve_array(inner, names, enclosing_namespace),
            Schema::Map(ref inner) => self.resolve_map(inner, names, enclosing_namespace),
            Schema::Record(RecordSchema { ref fields, .. }) => {
                self.resolve_record(fields, names, enclosing_namespace)
            }
            Schema::Decimal(DecimalSchema {
                scale,
                precision,
                ref inner,
            }) => self.resolve_decimal(precision, scale, inner),
            Schema::Date => self.resolve_date(),
            Schema::TimeMillis => self.resolve_time_millis(),
            Schema::TimeMicros => self.resolve_time_micros(),
            Schema::TimestampMillis => self.resolve_timestamp_millis(),
            Schema::TimestampMicros => self.resolve_timestamp_micros(),
            Schema::LocalTimestampMillis => self.resolve_local_timestamp_millis(),
            Schema::LocalTimestampMicros => self.resolve_local_timestamp_micros(),
            Schema::Duration => self.resolve_duration(),
            Schema::Uuid => self.resolve_uuid(),
        }
    }

    fn resolve_uuid(self) -> Result<Self, Error> {
        Ok(match self {
            uuid @ Value::Uuid(_) => uuid,
            Value::String(ref string) => {
                Value::Uuid(Uuid::from_str(string).map_err(Error::ConvertStrToUuid)?)
            }
            other => return Err(Error::GetUuid(other.into())),
        })
    }

    fn resolve_duration(self) -> Result<Self, Error> {
        Ok(match self {
            duration @ Value::Duration { .. } => duration,
            Value::Fixed(size, bytes) => {
                if size != 12 {
                    return Err(Error::GetDecimalFixedBytes(size));
                }
                Value::Duration(Duration::from([
                    bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
                    bytes[8], bytes[9], bytes[10], bytes[11],
                ]))
            }
            other => return Err(Error::ResolveDuration(other.into())),
        })
    }

    fn resolve_decimal(
        self,
        precision: Precision,
        scale: Scale,
        inner: &Schema,
    ) -> Result<Self, Error> {
        if scale > precision {
            return Err(Error::GetScaleAndPrecision { scale, precision });
        }
        match inner {
            &Schema::Fixed(FixedSchema { size, .. }) => {
                if max_prec_for_len(size)? < precision {
                    return Err(Error::GetScaleWithFixedSize { size, precision });
                }
            }
            Schema::Bytes => (),
            _ => return Err(Error::ResolveDecimalSchema(inner.into())),
        };
        match self {
            Value::Decimal(num) => {
                let num_bytes = num.len();
                if max_prec_for_len(num_bytes)? < precision {
                    Err(Error::ComparePrecisionAndSize {
                        precision,
                        num_bytes,
                    })
                } else {
                    Ok(Value::Decimal(num))
                }
                // check num.bits() here
            }
            Value::Fixed(_, bytes) | Value::Bytes(bytes) => {
                if max_prec_for_len(bytes.len())? < precision {
                    Err(Error::ComparePrecisionAndSize {
                        precision,
                        num_bytes: bytes.len(),
                    })
                } else {
                    // precision and scale match, can we assume the underlying type can hold the data?
                    Ok(Value::Decimal(Decimal::from(bytes)))
                }
            }
            other => Err(Error::ResolveDecimal(other.into())),
        }
    }

    fn resolve_date(self) -> Result<Self, Error> {
        match self {
            Value::Date(d) | Value::Int(d) => Ok(Value::Date(d)),
            other => Err(Error::GetDate(other.into())),
        }
    }

    fn resolve_time_millis(self) -> Result<Self, Error> {
        match self {
            Value::TimeMillis(t) | Value::Int(t) => Ok(Value::TimeMillis(t)),
            other => Err(Error::GetTimeMillis(other.into())),
        }
    }

    fn resolve_time_micros(self) -> Result<Self, Error> {
        match self {
            Value::TimeMicros(t) | Value::Long(t) => Ok(Value::TimeMicros(t)),
            Value::Int(t) => Ok(Value::TimeMicros(i64::from(t))),
            other => Err(Error::GetTimeMicros(other.into())),
        }
    }

    fn resolve_timestamp_millis(self) -> Result<Self, Error> {
        match self {
            Value::TimestampMillis(ts) | Value::Long(ts) => Ok(Value::TimestampMillis(ts)),
            Value::Int(ts) => Ok(Value::TimestampMillis(i64::from(ts))),
            other => Err(Error::GetTimestampMillis(other.into())),
        }
    }

    fn resolve_timestamp_micros(self) -> Result<Self, Error> {
        match self {
            Value::TimestampMicros(ts) | Value::Long(ts) => Ok(Value::TimestampMicros(ts)),
            Value::Int(ts) => Ok(Value::TimestampMicros(i64::from(ts))),
            other => Err(Error::GetTimestampMicros(other.into())),
        }
    }

    fn resolve_local_timestamp_millis(self) -> Result<Self, Error> {
        match self {
            Value::LocalTimestampMillis(ts) | Value::Long(ts) => {
                Ok(Value::LocalTimestampMillis(ts))
            }
            Value::Int(ts) => Ok(Value::LocalTimestampMillis(i64::from(ts))),
            other => Err(Error::GetLocalTimestampMillis(other.into())),
        }
    }

    fn resolve_local_timestamp_micros(self) -> Result<Self, Error> {
        match self {
            Value::LocalTimestampMicros(ts) | Value::Long(ts) => {
                Ok(Value::LocalTimestampMicros(ts))
            }
            Value::Int(ts) => Ok(Value::LocalTimestampMicros(i64::from(ts))),
            other => Err(Error::GetLocalTimestampMicros(other.into())),
        }
    }

    fn resolve_null(self) -> Result<Self, Error> {
        match self {
            Value::Null => Ok(Value::Null),
            other => Err(Error::GetNull(other.into())),
        }
    }

    fn resolve_boolean(self) -> Result<Self, Error> {
        match self {
            Value::Boolean(b) => Ok(Value::Boolean(b)),
            other => Err(Error::GetBoolean(other.into())),
        }
    }

    fn resolve_int(self) -> Result<Self, Error> {
        match self {
            Value::Int(n) => Ok(Value::Int(n)),
            Value::Long(n) => Ok(Value::Int(n as i32)),
            other => Err(Error::GetInt(other.into())),
        }
    }

    fn resolve_long(self) -> Result<Self, Error> {
        match self {
            Value::Int(n) => Ok(Value::Long(i64::from(n))),
            Value::Long(n) => Ok(Value::Long(n)),
            other => Err(Error::GetLong(other.into())),
        }
    }

    fn resolve_float(self) -> Result<Self, Error> {
        match self {
            Value::Int(n) => Ok(Value::Float(n as f32)),
            Value::Long(n) => Ok(Value::Float(n as f32)),
            Value::Float(x) => Ok(Value::Float(x)),
            Value::Double(x) => Ok(Value::Float(x as f32)),
            other => Err(Error::GetFloat(other.into())),
        }
    }

    fn resolve_double(self) -> Result<Self, Error> {
        match self {
            Value::Int(n) => Ok(Value::Double(f64::from(n))),
            Value::Long(n) => Ok(Value::Double(n as f64)),
            Value::Float(x) => Ok(Value::Double(f64::from(x))),
            Value::Double(x) => Ok(Value::Double(x)),
            other => Err(Error::GetDouble(other.into())),
        }
    }

    fn resolve_bytes(self) -> Result<Self, Error> {
        match self {
            Value::Bytes(bytes) => Ok(Value::Bytes(bytes)),
            Value::String(s) => Ok(Value::Bytes(s.into_bytes())),
            Value::Array(items) => Ok(Value::Bytes(
                items
                    .into_iter()
                    .map(Value::try_u8)
                    .collect::<Result<Vec<_>, _>>()?,
            )),
            other => Err(Error::GetBytes(other.into())),
        }
    }

    fn resolve_string(self) -> Result<Self, Error> {
        match self {
            Value::String(s) => Ok(Value::String(s)),
            Value::Bytes(bytes) | Value::Fixed(_, bytes) => Ok(Value::String(
                String::from_utf8(bytes).map_err(Error::ConvertToUtf8)?,
            )),
            other => Err(Error::GetString(other.into())),
        }
    }

    fn resolve_fixed(self, size: usize) -> Result<Self, Error> {
        match self {
            Value::Fixed(n, bytes) => {
                if n == size {
                    Ok(Value::Fixed(n, bytes))
                } else {
                    Err(Error::CompareFixedSizes { size, n })
                }
            }
            Value::String(s) => Ok(Value::Fixed(s.len(), s.into_bytes())),
            other => Err(Error::GetStringForFixed(other.into())),
        }
    }

    pub(crate) fn resolve_enum(
        self,
        symbols: &[String],
        enum_default: &Option<String>,
        _field_default: &Option<JsonValue>,
    ) -> Result<Self, Error> {
        let validate_symbol = |symbol: String, symbols: &[String]| {
            if let Some(index) = symbols.iter().position(|item| item == &symbol) {
                Ok(Value::Enum(index as u32, symbol))
            } else {
                match enum_default {
                    Some(default) => {
                        if let Some(index) = symbols.iter().position(|item| item == default) {
                            Ok(Value::Enum(index as u32, default.clone()))
                        } else {
                            Err(Error::GetEnumDefault {
                                symbol,
                                symbols: symbols.into(),
                            })
                        }
                    }
                    _ => Err(Error::GetEnumDefault {
                        symbol,
                        symbols: symbols.into(),
                    }),
                }
            }
        };

        match self {
            Value::Enum(_raw_index, s) => validate_symbol(s, symbols),
            Value::String(s) => validate_symbol(s, symbols),
            other => Err(Error::GetEnum(other.into())),
        }
    }

    fn resolve_union<S: Borrow<Schema> + Debug>(
        self,
        schema: &UnionSchema,
        names: &HashMap<Name, S>,
        enclosing_namespace: &Namespace,
        field_default: &Option<JsonValue>,
    ) -> Result<Self, Error> {
        let v = match self {
            // Both are unions case.
            Value::Union(_i, v) => *v,
            // Reader is a union, but writer is not.
            v => v,
        };
        let (i, inner) = schema
            .find_schema_with_known_schemata(&v, Some(names), enclosing_namespace)
            .ok_or(Error::FindUnionVariant)?;

        Ok(Value::Union(
            i as u32,
            Box::new(v.resolve_internal(inner, names, enclosing_namespace, field_default)?),
        ))
    }

    fn resolve_array<S: Borrow<Schema> + Debug>(
        self,
        schema: &Schema,
        names: &HashMap<Name, S>,
        enclosing_namespace: &Namespace,
    ) -> Result<Self, Error> {
        match self {
            Value::Array(items) => Ok(Value::Array(
                items
                    .into_iter()
                    .map(|item| item.resolve_internal(schema, names, enclosing_namespace, &None))
                    .collect::<Result<_, _>>()?,
            )),
            other => Err(Error::GetArray {
                expected: schema.into(),
                other: other.into(),
            }),
        }
    }

    fn resolve_map<S: Borrow<Schema> + Debug>(
        self,
        schema: &Schema,
        names: &HashMap<Name, S>,
        enclosing_namespace: &Namespace,
    ) -> Result<Self, Error> {
        match self {
            Value::Map(items) => Ok(Value::Map(
                items
                    .into_iter()
                    .map(|(key, value)| {
                        value
                            .resolve_internal(schema, names, enclosing_namespace, &None)
                            .map(|value| (key, value))
                    })
                    .collect::<Result<_, _>>()?,
            )),
            other => Err(Error::GetMap {
                expected: schema.into(),
                other: other.into(),
            }),
        }
    }

    fn resolve_record<S: Borrow<Schema> + Debug>(
        self,
        fields: &[RecordField],
        names: &HashMap<Name, S>,
        enclosing_namespace: &Namespace,
    ) -> Result<Self, Error> {
        let mut items = match self {
            Value::Map(items) => Ok(items),
            Value::Record(fields) => Ok(fields.into_iter().collect::<HashMap<_, _>>()),
            other => Err(Error::GetRecord {
                expected: fields
                    .iter()
                    .map(|field| (field.name.clone(), field.schema.clone().into()))
                    .collect(),
                other: other.into(),
            }),
        }?;

        let new_fields = fields
            .iter()
            .map(|field| {
                let value = match items.remove(&field.name) {
                    Some(value) => value,
                    None => match field.default {
                        Some(ref value) => match field.schema {
                            Schema::Enum(EnumSchema {
                                ref symbols,
                                ref default,
                                ..
                            }) => Value::from(value.clone()).resolve_enum(
                                symbols,
                                default,
                                &field.default.clone(),
                            )?,
                            Schema::Union(ref union_schema) => {
                                let first = &union_schema.variants()[0];
                                // NOTE: this match exists only to optimize null defaults for large
                                // backward-compatible schemas with many nullable fields
                                match first {
                                    Schema::Null => Value::Union(0, Box::new(Value::Null)),
                                    _ => Value::Union(
                                        0,
                                        Box::new(Value::from(value.clone()).resolve_internal(
                                            first,
                                            names,
                                            enclosing_namespace,
                                            &field.default,
                                        )?),
                                    ),
                                }
                            }
                            _ => Value::from(value.clone()),
                        },
                        None => {
                            return Err(Error::GetField(field.name.clone()));
                        }
                    },
                };
                value
                    .resolve_internal(&field.schema, names, enclosing_namespace, &field.default)
                    .map(|value| (field.name.clone(), value))
            })
            .collect::<Result<Vec<_>, _>>()?;

        Ok(Value::Record(new_fields))
    }

    fn try_u8(self) -> AvroResult<u8> {
        let int = self.resolve(&Schema::Int)?;
        if let Value::Int(n) = int {
            if n >= 0 && n <= i32::from(u8::MAX) {
                return Ok(n as u8);
            }
        }

        Err(Error::GetU8(int.into()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        decimal::Decimal,
        duration::{Days, Duration, Millis, Months},
        schema::{Name, RecordField, RecordFieldOrder, Schema, UnionSchema},
        types::Value,
    };
    use apache_avro_test_helper::{
        logger::{assert_logged, assert_not_logged},
        TestResult,
    };
    use num_bigint::BigInt;
    use pretty_assertions::assert_eq;
    use uuid::Uuid;

    #[test]
    fn avro_3809_validate_nested_records_with_implicit_namespace() -> TestResult {
        let schema = Schema::parse_str(
            r#"{
            "name": "record_name",
            "namespace": "space",
            "type": "record",
            "fields": [
              {
                "name": "outer_field_1",
                "type": {
                  "type": "record",
                  "name": "middle_record_name",
                  "namespace": "middle_namespace",
                  "fields": [
                    {
                      "name": "middle_field_1",
                      "type": {
                        "type": "record",
                        "name": "inner_record_name",
                        "fields": [
                          { "name": "inner_field_1", "type": "double" }
                        ]
                      }
                    },
                    { "name": "middle_field_2", "type": "inner_record_name" }
                  ]
                }
              }
            ]
          }"#,
        )?;
        let value = Value::Record(vec![(
            "outer_field_1".into(),
            Value::Record(vec![
                (
                    "middle_field_1".into(),
                    Value::Record(vec![("inner_field_1".into(), Value::Double(1.2f64))]),
                ),
                (
                    "middle_field_2".into(),
                    Value::Record(vec![("inner_field_1".into(), Value::Double(1.6f64))]),
                ),
            ]),
        )]);

        assert!(value.validate(&schema));
        Ok(())
    }

    #[test]
    fn validate() -> TestResult {
        let value_schema_valid = vec![
            (Value::Int(42), Schema::Int, true, ""),
            (Value::Int(43), Schema::Long, true, ""),
            (Value::Float(43.2), Schema::Float, true, ""),
            (Value::Float(45.9), Schema::Double, true, ""),
            (
                Value::Int(42),
                Schema::Boolean,
                false,
                "Invalid value: Int(42) for schema: Boolean. Reason: Unsupported value-schema combination",
            ),
            (
                Value::Union(0, Box::new(Value::Null)),
                Schema::Union(UnionSchema::new(vec![Schema::Null, Schema::Int])?),
                true,
                "",
            ),
            (
                Value::Union(1, Box::new(Value::Int(42))),
                Schema::Union(UnionSchema::new(vec![Schema::Null, Schema::Int])?),
                true,
                "",
            ),
            (
                Value::Union(0, Box::new(Value::Null)),
                Schema::Union(UnionSchema::new(vec![Schema::Double, Schema::Int])?),
                false,
                "Invalid value: Union(0, Null) for schema: Union(UnionSchema { schemas: [Double, Int], variant_index: {Int: 1, Double: 0} }). Reason: Unsupported value-schema combination",
            ),
            (
                Value::Union(3, Box::new(Value::Int(42))),
                Schema::Union(
                    UnionSchema::new(vec![
                        Schema::Null,
                        Schema::Double,
                        Schema::String,
                        Schema::Int,
                    ])
                    ?,
                ),
                true,
                "",
            ),
            (
                Value::Union(1, Box::new(Value::Long(42i64))),
                Schema::Union(
                    UnionSchema::new(vec![Schema::Null, Schema::TimestampMillis])?,
                ),
                true,
                "",
            ),
            (
                Value::Union(2, Box::new(Value::Long(1_i64))),
                Schema::Union(UnionSchema::new(vec![Schema::Null, Schema::Int])?),
                false,
                "Invalid value: Union(2, Long(1)) for schema: Union(UnionSchema { schemas: [Null, Int], variant_index: {Null: 0, Int: 1} }). Reason: No schema in the union at position '2'",
            ),
            (
                Value::Array(vec![Value::Long(42i64)]),
                Schema::Array(Box::new(Schema::Long)),
                true,
                "",
            ),
            (
                Value::Array(vec![Value::Boolean(true)]),
                Schema::Array(Box::new(Schema::Long)),
                false,
                "Invalid value: Array([Boolean(true)]) for schema: Array(Long). Reason: Unsupported value-schema combination",
            ),
            (Value::Record(vec![]), Schema::Null, false, "Invalid value: Record([]) for schema: Null. Reason: Unsupported value-schema combination"),
            (
                Value::Fixed(12, vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]),
                Schema::Duration,
                true,
                "",
            ),
            (
                Value::Fixed(11, vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]),
                Schema::Duration,
                false,
                "Invalid value: Fixed(11, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) for schema: Duration. Reason: The value's size ('11') must be exactly 12 to be a Duration",
            ),
            (
                Value::Record(vec![("unknown_field_name".to_string(), Value::Null)]),
                Schema::Record(RecordSchema {
                    name: Name::new("record_name").unwrap(),
                    aliases: None,
                    doc: None,
                    fields: vec![RecordField {
                        name: "field_name".to_string(),
                        doc: None,
                        default: None,
                        aliases: None,
                        schema: Schema::Int,
                        order: RecordFieldOrder::Ignore,
                        position: 0,
                        custom_attributes: Default::default(),
                    }],
                    lookup: Default::default(),
                    attributes: Default::default(),
                }),
                false,
                r#"Invalid value: Record([("unknown_field_name", Null)]) for schema: Record(RecordSchema { name: Name { name: "record_name", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "field_name", doc: None, aliases: None, default: None, schema: Int, order: Ignore, position: 0, custom_attributes: {} }], lookup: {}, attributes: {} }). Reason: There is no schema field for field 'unknown_field_name'"#,
            ),
            (
                Value::Record(vec![("field_name".to_string(), Value::Null)]),
                Schema::Record(RecordSchema {
                    name: Name::new("record_name").unwrap(),
                    aliases: None,
                    doc: None,
                    fields: vec![RecordField {
                        name: "field_name".to_string(),
                        doc: None,
                        default: None,
                        aliases: None,
                        schema: Schema::Ref {
                            name: Name::new("missing").unwrap(),
                        },
                        order: RecordFieldOrder::Ignore,
                        position: 0,
                        custom_attributes: Default::default(),
                    }],
                    lookup: [("field_name".to_string(), 0)].iter().cloned().collect(),
                    attributes: Default::default(),
                }),
                false,
                r#"Invalid value: Record([("field_name", Null)]) for schema: Record(RecordSchema { name: Name { name: "record_name", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "field_name", doc: None, aliases: None, default: None, schema: Ref { name: Name { name: "missing", namespace: None } }, order: Ignore, position: 0, custom_attributes: {} }], lookup: {"field_name": 0}, attributes: {} }). Reason: Unresolved schema reference: 'Name { name: "missing", namespace: None }'. Parsed names: []"#,
            ),
        ];

        for (value, schema, valid, expected_err_message) in value_schema_valid.into_iter() {
            let err_message =
                value.validate_internal::<Schema>(&schema, &HashMap::default(), &None);
            assert_eq!(valid, err_message.is_none());
            if !valid {
                let full_err_message = format!(
                    "Invalid value: {:?} for schema: {:?}. Reason: {}",
                    value,
                    schema,
                    err_message.unwrap()
                );
                assert_eq!(expected_err_message, full_err_message);
            }
        }

        Ok(())
    }

    #[test]
    fn validate_fixed() -> TestResult {
        let schema = Schema::Fixed(FixedSchema {
            size: 4,
            name: Name::new("some_fixed").unwrap(),
            aliases: None,
            doc: None,
            attributes: Default::default(),
        });

        assert!(Value::Fixed(4, vec![0, 0, 0, 0]).validate(&schema));
        let value = Value::Fixed(5, vec![0, 0, 0, 0, 0]);
        assert!(!value.validate(&schema));
        assert_logged(
            format!(
                "Invalid value: {:?} for schema: {:?}. Reason: {}",
                value, schema, "The value's size (5) is different than the schema's size (4)"
            )
            .as_str(),
        );

        assert!(Value::Bytes(vec![0, 0, 0, 0]).validate(&schema));
        let value = Value::Bytes(vec![0, 0, 0, 0, 0]);
        assert!(!value.validate(&schema));
        assert_logged(
            format!(
                "Invalid value: {:?} for schema: {:?}. Reason: {}",
                value, schema, "The bytes' length (5) is different than the schema's size (4)"
            )
            .as_str(),
        );

        Ok(())
    }

    #[test]
    fn validate_enum() -> TestResult {
        let schema = Schema::Enum(EnumSchema {
            name: Name::new("some_enum").unwrap(),
            aliases: None,
            doc: None,
            symbols: vec![
                "spades".to_string(),
                "hearts".to_string(),
                "diamonds".to_string(),
                "clubs".to_string(),
            ],
            default: None,
            attributes: Default::default(),
        });

        assert!(Value::Enum(0, "spades".to_string()).validate(&schema));
        assert!(Value::String("spades".to_string()).validate(&schema));

        let value = Value::Enum(1, "spades".to_string());
        assert!(!value.validate(&schema));
        assert_logged(
            format!(
                "Invalid value: {:?} for schema: {:?}. Reason: {}",
                value, schema, "Symbol 'spades' is not at position '1'"
            )
            .as_str(),
        );

        let value = Value::Enum(1000, "spades".to_string());
        assert!(!value.validate(&schema));
        assert_logged(
            format!(
                "Invalid value: {:?} for schema: {:?}. Reason: {}",
                value, schema, "No symbol at position '1000'"
            )
            .as_str(),
        );

        let value = Value::String("lorem".to_string());
        assert!(!value.validate(&schema));
        assert_logged(
            format!(
                "Invalid value: {:?} for schema: {:?}. Reason: {}",
                value, schema, "'lorem' is not a member of the possible symbols"
            )
            .as_str(),
        );

        let other_schema = Schema::Enum(EnumSchema {
            name: Name::new("some_other_enum").unwrap(),
            aliases: None,
            doc: None,
            symbols: vec![
                "hearts".to_string(),
                "diamonds".to_string(),
                "clubs".to_string(),
                "spades".to_string(),
            ],
            default: None,
            attributes: Default::default(),
        });

        let value = Value::Enum(0, "spades".to_string());
        assert!(!value.validate(&other_schema));
        assert_logged(
            format!(
                "Invalid value: {:?} for schema: {:?}. Reason: {}",
                value, other_schema, "Symbol 'spades' is not at position '0'"
            )
            .as_str(),
        );

        Ok(())
    }

    #[test]
    fn validate_record() -> TestResult {
        // {
        //    "type": "record",
        //    "fields": [
        //      {"type": "long", "name": "a"},
        //      {"type": "string", "name": "b"},
        //      {
        //          "type": ["null", "int"]
        //          "name": "c",
        //          "default": null
        //      }
        //    ]
        // }
        let schema = Schema::Record(RecordSchema {
            name: Name::new("some_record").unwrap(),
            aliases: None,
            doc: None,
            fields: vec![
                RecordField {
                    name: "a".to_string(),
                    doc: None,
                    default: None,
                    aliases: None,
                    schema: Schema::Long,
                    order: RecordFieldOrder::Ascending,
                    position: 0,
                    custom_attributes: Default::default(),
                },
                RecordField {
                    name: "b".to_string(),
                    doc: None,
                    default: None,
                    aliases: None,
                    schema: Schema::String,
                    order: RecordFieldOrder::Ascending,
                    position: 1,
                    custom_attributes: Default::default(),
                },
                RecordField {
                    name: "c".to_string(),
                    doc: None,
                    default: Some(JsonValue::Null),
                    aliases: None,
                    schema: Schema::Union(UnionSchema::new(vec![Schema::Null, Schema::Int])?),
                    order: RecordFieldOrder::Ascending,
                    position: 2,
                    custom_attributes: Default::default(),
                },
            ],
            lookup: [
                ("a".to_string(), 0),
                ("b".to_string(), 1),
                ("c".to_string(), 2),
            ]
            .iter()
            .cloned()
            .collect(),
            attributes: Default::default(),
        });

        assert!(Value::Record(vec![
            ("a".to_string(), Value::Long(42i64)),
            ("b".to_string(), Value::String("foo".to_string())),
        ])
        .validate(&schema));

        let value = Value::Record(vec![
            ("b".to_string(), Value::String("foo".to_string())),
            ("a".to_string(), Value::Long(42i64)),
        ]);
        assert!(value.validate(&schema));

        let value = Value::Record(vec![
            ("a".to_string(), Value::Boolean(false)),
            ("b".to_string(), Value::String("foo".to_string())),
        ]);
        assert!(!value.validate(&schema));
        assert_logged(
            r#"Invalid value: Record([("a", Boolean(false)), ("b", String("foo"))]) for schema: Record(RecordSchema { name: Name { name: "some_record", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "a", doc: None, aliases: None, default: None, schema: Long, order: Ascending, position: 0, custom_attributes: {} }, RecordField { name: "b", doc: None, aliases: None, default: None, schema: String, order: Ascending, position: 1, custom_attributes: {} }, RecordField { name: "c", doc: None, aliases: None, default: Some(Null), schema: Union(UnionSchema { schemas: [Null, Int], variant_index: {Null: 0, Int: 1} }), order: Ascending, position: 2, custom_attributes: {} }], lookup: {"a": 0, "b": 1, "c": 2}, attributes: {} }). Reason: Unsupported value-schema combination"#,
        );

        let value = Value::Record(vec![
            ("a".to_string(), Value::Long(42i64)),
            ("c".to_string(), Value::String("foo".to_string())),
        ]);
        assert!(!value.validate(&schema));
        assert_logged(
            r#"Invalid value: Record([("a", Long(42)), ("c", String("foo"))]) for schema: Record(RecordSchema { name: Name { name: "some_record", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "a", doc: None, aliases: None, default: None, schema: Long, order: Ascending, position: 0, custom_attributes: {} }, RecordField { name: "b", doc: None, aliases: None, default: None, schema: String, order: Ascending, position: 1, custom_attributes: {} }, RecordField { name: "c", doc: None, aliases: None, default: Some(Null), schema: Union(UnionSchema { schemas: [Null, Int], variant_index: {Null: 0, Int: 1} }), order: Ascending, position: 2, custom_attributes: {} }], lookup: {"a": 0, "b": 1, "c": 2}, attributes: {} }). Reason: Could not find matching type in union"#,
        );
        assert_not_logged(
            r#"Invalid value: String("foo") for schema: Int. Reason: Unsupported value-schema combination"#,
        );

        let value = Value::Record(vec![
            ("a".to_string(), Value::Long(42i64)),
            ("d".to_string(), Value::String("foo".to_string())),
        ]);
        assert!(!value.validate(&schema));
        assert_logged(
            r#"Invalid value: Record([("a", Long(42)), ("d", String("foo"))]) for schema: Record(RecordSchema { name: Name { name: "some_record", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "a", doc: None, aliases: None, default: None, schema: Long, order: Ascending, position: 0, custom_attributes: {} }, RecordField { name: "b", doc: None, aliases: None, default: None, schema: String, order: Ascending, position: 1, custom_attributes: {} }, RecordField { name: "c", doc: None, aliases: None, default: Some(Null), schema: Union(UnionSchema { schemas: [Null, Int], variant_index: {Null: 0, Int: 1} }), order: Ascending, position: 2, custom_attributes: {} }], lookup: {"a": 0, "b": 1, "c": 2}, attributes: {} }). Reason: There is no schema field for field 'd'"#,
        );

        let value = Value::Record(vec![
            ("a".to_string(), Value::Long(42i64)),
            ("b".to_string(), Value::String("foo".to_string())),
            ("c".to_string(), Value::Null),
            ("d".to_string(), Value::Null),
        ]);
        assert!(!value.validate(&schema));
        assert_logged(
            r#"Invalid value: Record([("a", Long(42)), ("b", String("foo")), ("c", Null), ("d", Null)]) for schema: Record(RecordSchema { name: Name { name: "some_record", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "a", doc: None, aliases: None, default: None, schema: Long, order: Ascending, position: 0, custom_attributes: {} }, RecordField { name: "b", doc: None, aliases: None, default: None, schema: String, order: Ascending, position: 1, custom_attributes: {} }, RecordField { name: "c", doc: None, aliases: None, default: Some(Null), schema: Union(UnionSchema { schemas: [Null, Int], variant_index: {Null: 0, Int: 1} }), order: Ascending, position: 2, custom_attributes: {} }], lookup: {"a": 0, "b": 1, "c": 2}, attributes: {} }). Reason: The value's records length (4) is greater than the schema's (3 fields)"#,
        );

        assert!(Value::Map(
            vec![
                ("a".to_string(), Value::Long(42i64)),
                ("b".to_string(), Value::String("foo".to_string())),
            ]
            .into_iter()
            .collect()
        )
        .validate(&schema));

        assert!(!Value::Map(
            vec![("d".to_string(), Value::Long(123_i64)),]
                .into_iter()
                .collect()
        )
        .validate(&schema));
        assert_logged(
            r#"Invalid value: Map({"d": Long(123)}) for schema: Record(RecordSchema { name: Name { name: "some_record", namespace: None }, aliases: None, doc: None, fields: [RecordField { name: "a", doc: None, aliases: None, default: None, schema: Long, order: Ascending, position: 0, custom_attributes: {} }, RecordField { name: "b", doc: None, aliases: None, default: None, schema: String, order: Ascending, position: 1, custom_attributes: {} }, RecordField { name: "c", doc: None, aliases: None, default: Some(Null), schema: Union(UnionSchema { schemas: [Null, Int], variant_index: {Null: 0, Int: 1} }), order: Ascending, position: 2, custom_attributes: {} }], lookup: {"a": 0, "b": 1, "c": 2}, attributes: {} }). Reason: Field with name '"a"' is not a member of the map items
Field with name '"b"' is not a member of the map items"#,
        );

        let union_schema = Schema::Union(UnionSchema::new(vec![Schema::Null, schema])?);

        assert!(Value::Union(
            1,
            Box::new(Value::Record(vec![
                ("a".to_string(), Value::Long(42i64)),
                ("b".to_string(), Value::String("foo".to_string())),
            ]))
        )
        .validate(&union_schema));

        assert!(Value::Union(
            1,
            Box::new(Value::Map(
                vec![
                    ("a".to_string(), Value::Long(42i64)),
                    ("b".to_string(), Value::String("foo".to_string())),
                ]
                .into_iter()
                .collect()
            ))
        )
        .validate(&union_schema));

        Ok(())
    }

    #[test]
    fn resolve_bytes_ok() -> TestResult {
        let value = Value::Array(vec![Value::Int(0), Value::Int(42)]);
        assert_eq!(
            value.resolve(&Schema::Bytes)?,
            Value::Bytes(vec![0u8, 42u8])
        );

        Ok(())
    }

    #[test]
    fn resolve_string_from_bytes() -> TestResult {
        let value = Value::Bytes(vec![97, 98, 99]);
        assert_eq!(
            value.resolve(&Schema::String)?,
            Value::String("abc".to_string())
        );

        Ok(())
    }

    #[test]
    fn resolve_string_from_fixed() -> TestResult {
        let value = Value::Fixed(3, vec![97, 98, 99]);
        assert_eq!(
            value.resolve(&Schema::String)?,
            Value::String("abc".to_string())
        );

        Ok(())
    }

    #[test]
    fn resolve_bytes_failure() {
        let value = Value::Array(vec![Value::Int(2000), Value::Int(-42)]);
        assert!(value.resolve(&Schema::Bytes).is_err());
    }

    #[test]
    fn resolve_decimal_bytes() -> TestResult {
        let value = Value::Decimal(Decimal::from(vec![1, 2, 3, 4, 5]));
        value.clone().resolve(&Schema::Decimal(DecimalSchema {
            precision: 10,
            scale: 4,
            inner: Box::new(Schema::Bytes),
        }))?;
        assert!(value.resolve(&Schema::String).is_err());

        Ok(())
    }

    #[test]
    fn resolve_decimal_invalid_scale() {
        let value = Value::Decimal(Decimal::from(vec![1, 2]));
        assert!(value
            .resolve(&Schema::Decimal(DecimalSchema {
                precision: 2,
                scale: 3,
                inner: Box::new(Schema::Bytes),
            }))
            .is_err());
    }

    #[test]
    fn resolve_decimal_invalid_precision_for_length() {
        let value = Value::Decimal(Decimal::from((1u8..=8u8).rev().collect::<Vec<_>>()));
        assert!(value
            .resolve(&Schema::Decimal(DecimalSchema {
                precision: 1,
                scale: 0,
                inner: Box::new(Schema::Bytes),
            }))
            .is_ok());
    }

    #[test]
    fn resolve_decimal_fixed() {
        let value = Value::Decimal(Decimal::from(vec![1, 2, 3, 4, 5]));
        assert!(value
            .clone()
            .resolve(&Schema::Decimal(DecimalSchema {
                precision: 10,
                scale: 1,
                inner: Box::new(Schema::Fixed(FixedSchema {
                    name: Name::new("decimal").unwrap(),
                    aliases: None,
                    size: 20,
                    doc: None,
                    attributes: Default::default(),
                }))
            }))
            .is_ok());
        assert!(value.resolve(&Schema::String).is_err());
    }

    #[test]
    fn resolve_date() {
        let value = Value::Date(2345);
        assert!(value.clone().resolve(&Schema::Date).is_ok());
        assert!(value.resolve(&Schema::String).is_err());
    }

    #[test]
    fn resolve_time_millis() {
        let value = Value::TimeMillis(10);
        assert!(value.clone().resolve(&Schema::TimeMillis).is_ok());
        assert!(value.resolve(&Schema::TimeMicros).is_err());
    }

    #[test]
    fn resolve_time_micros() {
        let value = Value::TimeMicros(10);
        assert!(value.clone().resolve(&Schema::TimeMicros).is_ok());
        assert!(value.resolve(&Schema::TimeMillis).is_err());
    }

    #[test]
    fn resolve_timestamp_millis() {
        let value = Value::TimestampMillis(10);
        assert!(value.clone().resolve(&Schema::TimestampMillis).is_ok());
        assert!(value.resolve(&Schema::Float).is_err());

        let value = Value::Float(10.0f32);
        assert!(value.resolve(&Schema::TimestampMillis).is_err());
    }

    #[test]
    fn resolve_timestamp_micros() {
        let value = Value::TimestampMicros(10);
        assert!(value.clone().resolve(&Schema::TimestampMicros).is_ok());
        assert!(value.resolve(&Schema::Int).is_err());

        let value = Value::Double(10.0);
        assert!(value.resolve(&Schema::TimestampMicros).is_err());
    }

    #[test]
    fn test_avro_3853_resolve_timestamp_millis() {
        let value = Value::LocalTimestampMillis(10);
        assert!(value.clone().resolve(&Schema::LocalTimestampMillis).is_ok());
        assert!(value.resolve(&Schema::Float).is_err());

        let value = Value::Float(10.0f32);
        assert!(value.resolve(&Schema::LocalTimestampMillis).is_err());
    }

    #[test]
    fn test_avro_3853_resolve_timestamp_micros() {
        let value = Value::LocalTimestampMicros(10);
        assert!(value.clone().resolve(&Schema::LocalTimestampMicros).is_ok());
        assert!(value.resolve(&Schema::Int).is_err());

        let value = Value::Double(10.0);
        assert!(value.resolve(&Schema::LocalTimestampMicros).is_err());
    }

    #[test]
    fn resolve_duration() {
        let value = Value::Duration(Duration::new(
            Months::new(10),
            Days::new(5),
            Millis::new(3000),
        ));
        assert!(value.clone().resolve(&Schema::Duration).is_ok());
        assert!(value.resolve(&Schema::TimestampMicros).is_err());
        assert!(Value::Long(1i64).resolve(&Schema::Duration).is_err());
    }

    #[test]
    fn resolve_uuid() -> TestResult {
        let value = Value::Uuid(Uuid::parse_str("1481531d-ccc9-46d9-a56f-5b67459c0537")?);
        assert!(value.clone().resolve(&Schema::Uuid).is_ok());
        assert!(value.resolve(&Schema::TimestampMicros).is_err());

        Ok(())
    }

    #[test]
    fn avro_3678_resolve_float_to_double() {
        let value = Value::Float(2345.1);
        assert!(value.resolve(&Schema::Double).is_ok());
    }

    #[test]
    fn test_avro_3621_resolve_to_nullable_union() -> TestResult {
        let schema = Schema::parse_str(
            r#"{
            "type": "record",
            "name": "root",
            "fields": [
                {
                    "name": "event",
                    "type": [
                        "null", 
                        {
                            "type": "record",
                            "name": "event",
                            "fields": [
                                {
                                    "name": "amount",
                                    "type": "int"
                                },
                                {
                                    "name": "size",
                                    "type": [
                                        "null",
                                        "int"
                                    ],
                                    "default": null
                                }
                            ]
                        }
                    ],
                    "default": null
                }
            ]
        }"#,
        )?;

        let value = Value::Record(vec![(
            "event".to_string(),
            Value::Record(vec![("amount".to_string(), Value::Int(200))]),
        )]);
        assert!(value.resolve(&schema).is_ok());

        let value = Value::Record(vec![(
            "event".to_string(),
            Value::Record(vec![("size".to_string(), Value::Int(1))]),
        )]);
        assert!(value.resolve(&schema).is_err());

        Ok(())
    }

    #[test]
    fn json_from_avro() -> TestResult {
        assert_eq!(JsonValue::try_from(Value::Null)?, JsonValue::Null);
        assert_eq!(
            JsonValue::try_from(Value::Boolean(true))?,
            JsonValue::Bool(true)
        );
        assert_eq!(
            JsonValue::try_from(Value::Int(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::Long(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::Float(1.0))?,
            JsonValue::Number(Number::from_f64(1.0).unwrap())
        );
        assert_eq!(
            JsonValue::try_from(Value::Double(1.0))?,
            JsonValue::Number(Number::from_f64(1.0).unwrap())
        );
        assert_eq!(
            JsonValue::try_from(Value::Bytes(vec![1, 2, 3]))?,
            JsonValue::Array(vec![
                JsonValue::Number(1.into()),
                JsonValue::Number(2.into()),
                JsonValue::Number(3.into())
            ])
        );
        assert_eq!(
            JsonValue::try_from(Value::String("test".into()))?,
            JsonValue::String("test".into())
        );
        assert_eq!(
            JsonValue::try_from(Value::Fixed(3, vec![1, 2, 3]))?,
            JsonValue::Array(vec![
                JsonValue::Number(1.into()),
                JsonValue::Number(2.into()),
                JsonValue::Number(3.into())
            ])
        );
        assert_eq!(
            JsonValue::try_from(Value::Enum(1, "test_enum".into()))?,
            JsonValue::String("test_enum".into())
        );
        assert_eq!(
            JsonValue::try_from(Value::Union(1, Box::new(Value::String("test_enum".into()))))?,
            JsonValue::String("test_enum".into())
        );
        assert_eq!(
            JsonValue::try_from(Value::Array(vec![
                Value::Int(1),
                Value::Int(2),
                Value::Int(3)
            ]))?,
            JsonValue::Array(vec![
                JsonValue::Number(1.into()),
                JsonValue::Number(2.into()),
                JsonValue::Number(3.into())
            ])
        );
        assert_eq!(
            JsonValue::try_from(Value::Map(
                vec![
                    ("v1".to_string(), Value::Int(1)),
                    ("v2".to_string(), Value::Int(2)),
                    ("v3".to_string(), Value::Int(3))
                ]
                .into_iter()
                .collect()
            ))?,
            JsonValue::Object(
                vec![
                    ("v1".to_string(), JsonValue::Number(1.into())),
                    ("v2".to_string(), JsonValue::Number(2.into())),
                    ("v3".to_string(), JsonValue::Number(3.into()))
                ]
                .into_iter()
                .collect()
            )
        );
        assert_eq!(
            JsonValue::try_from(Value::Record(vec![
                ("v1".to_string(), Value::Int(1)),
                ("v2".to_string(), Value::Int(2)),
                ("v3".to_string(), Value::Int(3))
            ]))?,
            JsonValue::Object(
                vec![
                    ("v1".to_string(), JsonValue::Number(1.into())),
                    ("v2".to_string(), JsonValue::Number(2.into())),
                    ("v3".to_string(), JsonValue::Number(3.into()))
                ]
                .into_iter()
                .collect()
            )
        );
        assert_eq!(
            JsonValue::try_from(Value::Date(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::Decimal(vec![1, 2, 3].into()))?,
            JsonValue::Array(vec![
                JsonValue::Number(1.into()),
                JsonValue::Number(2.into()),
                JsonValue::Number(3.into())
            ])
        );
        assert_eq!(
            JsonValue::try_from(Value::TimeMillis(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::TimeMicros(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::TimestampMillis(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::TimestampMicros(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::LocalTimestampMillis(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::LocalTimestampMicros(1))?,
            JsonValue::Number(1.into())
        );
        assert_eq!(
            JsonValue::try_from(Value::Duration(
                [1u8, 2u8, 3u8, 4u8, 5u8, 6u8, 7u8, 8u8, 9u8, 10u8, 11u8, 12u8].into()
            ))?,
            JsonValue::Array(vec![
                JsonValue::Number(1.into()),
                JsonValue::Number(2.into()),
                JsonValue::Number(3.into()),
                JsonValue::Number(4.into()),
                JsonValue::Number(5.into()),
                JsonValue::Number(6.into()),
                JsonValue::Number(7.into()),
                JsonValue::Number(8.into()),
                JsonValue::Number(9.into()),
                JsonValue::Number(10.into()),
                JsonValue::Number(11.into()),
                JsonValue::Number(12.into()),
            ])
        );
        assert_eq!(
            JsonValue::try_from(Value::Uuid(Uuid::parse_str(
                "936DA01F-9ABD-4D9D-80C7-02AF85C822A8"
            )?))?,
            JsonValue::String("936da01f-9abd-4d9d-80c7-02af85c822a8".into())
        );

        Ok(())
    }

    #[test]
    fn test_avro_3433_recursive_resolves_record() -> TestResult {
        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":{
                        "type":"record",
                        "name": "Inner",
                        "fields": [ {
                            "name":"z",
                            "type":"int"
                        }]
                    }
                },
                {
                    "name":"b",
                    "type":"Inner"
                }
            ]
        }"#,
        )?;

        let inner_value1 = Value::Record(vec![("z".into(), Value::Int(3))]);
        let inner_value2 = Value::Record(vec![("z".into(), Value::Int(6))]);
        let outer = Value::Record(vec![("a".into(), inner_value1), ("b".into(), inner_value2)]);
        outer
            .resolve(&schema)
            .expect("Record definition defined in one field must be available in other field");

        Ok(())
    }

    #[test]
    fn test_avro_3433_recursive_resolves_array() -> TestResult {
        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":{
                        "type":"array",
                        "items": {
                            "type":"record",
                            "name": "Inner",
                            "fields": [ {
                                "name":"z",
                                "type":"int"
                            }]
                        }
                    }
                },
                {
                    "name":"b",
                    "type": {
                        "type":"map",
                        "values":"Inner"
                    }
                }
            ]
        }"#,
        )?;

        let inner_value1 = Value::Record(vec![("z".into(), Value::Int(3))]);
        let inner_value2 = Value::Record(vec![("z".into(), Value::Int(6))]);
        let outer_value = Value::Record(vec![
            ("a".into(), Value::Array(vec![inner_value1])),
            (
                "b".into(),
                Value::Map(vec![("akey".into(), inner_value2)].into_iter().collect()),
            ),
        ]);
        outer_value
            .resolve(&schema)
            .expect("Record defined in array definition must be resolvable from map");

        Ok(())
    }

    #[test]
    fn test_avro_3433_recursive_resolves_map() -> TestResult {
        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":{
                        "type":"record",
                        "name": "Inner",
                        "fields": [ {
                            "name":"z",
                            "type":"int"
                        }]
                    }
                },
                {
                    "name":"b",
                    "type": {
                        "type":"map",
                        "values":"Inner"
                    }
                }
            ]
        }"#,
        )?;

        let inner_value1 = Value::Record(vec![("z".into(), Value::Int(3))]);
        let inner_value2 = Value::Record(vec![("z".into(), Value::Int(6))]);
        let outer_value = Value::Record(vec![
            ("a".into(), inner_value1),
            (
                "b".into(),
                Value::Map(vec![("akey".into(), inner_value2)].into_iter().collect()),
            ),
        ]);
        outer_value
            .resolve(&schema)
            .expect("Record defined in record field must be resolvable from map field");

        Ok(())
    }

    #[test]
    fn test_avro_3433_recursive_resolves_record_wrapper() -> TestResult {
        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":{
                        "type":"record",
                        "name": "Inner",
                        "fields": [ {
                            "name":"z",
                            "type":"int"
                        }]
                    }
                },
                {
                    "name":"b",
                    "type": {
                        "type":"record",
                        "name": "InnerWrapper",
                        "fields": [ {
                            "name":"j",
                            "type":"Inner"
                        }]
                    }
                }
            ]
        }"#,
        )?;

        let inner_value1 = Value::Record(vec![("z".into(), Value::Int(3))]);
        let inner_value2 = Value::Record(vec![(
            "j".into(),
            Value::Record(vec![("z".into(), Value::Int(6))]),
        )]);
        let outer_value =
            Value::Record(vec![("a".into(), inner_value1), ("b".into(), inner_value2)]);
        outer_value.resolve(&schema).expect("Record schema defined in field must be resolvable in Record schema defined in other field");

        Ok(())
    }

    #[test]
    fn test_avro_3433_recursive_resolves_map_and_array() -> TestResult {
        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":{
                        "type":"map",
                        "values": {
                            "type":"record",
                            "name": "Inner",
                            "fields": [ {
                                "name":"z",
                                "type":"int"
                            }]
                        }
                    }
                },
                {
                    "name":"b",
                    "type": {
                        "type":"array",
                        "items":"Inner"
                    }
                }
            ]
        }"#,
        )?;

        let inner_value1 = Value::Record(vec![("z".into(), Value::Int(3))]);
        let inner_value2 = Value::Record(vec![("z".into(), Value::Int(6))]);
        let outer_value = Value::Record(vec![
            (
                "a".into(),
                Value::Map(vec![("akey".into(), inner_value2)].into_iter().collect()),
            ),
            ("b".into(), Value::Array(vec![inner_value1])),
        ]);
        outer_value
            .resolve(&schema)
            .expect("Record defined in map definition must be resolvable from array");

        Ok(())
    }

    #[test]
    fn test_avro_3433_recursive_resolves_union() -> TestResult {
        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":["null", {
                        "type":"record",
                        "name": "Inner",
                        "fields": [ {
                            "name":"z",
                            "type":"int"
                        }]
                    }]
                },
                {
                    "name":"b",
                    "type":"Inner"
                }
            ]
        }"#,
        )?;

        let inner_value1 = Value::Record(vec![("z".into(), Value::Int(3))]);
        let inner_value2 = Value::Record(vec![("z".into(), Value::Int(6))]);
        let outer1 = Value::Record(vec![
            ("a".into(), inner_value1),
            ("b".into(), inner_value2.clone()),
        ]);
        outer1
            .resolve(&schema)
            .expect("Record definition defined in union must be resolved in other field");
        let outer2 = Value::Record(vec![("a".into(), Value::Null), ("b".into(), inner_value2)]);
        outer2
            .resolve(&schema)
            .expect("Record definition defined in union must be resolved in other field");

        Ok(())
    }

    #[test]
    fn test_avro_3461_test_multi_level_resolve_outer_namespace() -> TestResult {
        let schema = r#"
        {
          "name": "record_name",
          "namespace": "space",
          "type": "record",
          "fields": [
            {
              "name": "outer_field_1",
              "type": [
                        "null",
                        {
                            "type": "record",
                            "name": "middle_record_name",
                            "fields":[
                                {
                                    "name":"middle_field_1",
                                    "type":[
                                        "null",
                                        {
                                            "type":"record",
                                            "name":"inner_record_name",
                                            "fields":[
                                                {
                                                    "name":"inner_field_1",
                                                    "type":"double"
                                                }
                                            ]
                                        }
                                    ]
                                }
                            ]
                        }
                    ]
            },
            {
                "name": "outer_field_2",
                "type" : "space.inner_record_name"
            }
          ]
        }
        "#;
        let schema = Schema::parse_str(schema)?;
        let inner_record = Value::Record(vec![("inner_field_1".into(), Value::Double(5.4))]);
        let middle_record_variation_1 = Value::Record(vec![(
            "middle_field_1".into(),
            Value::Union(0, Box::new(Value::Null)),
        )]);
        let middle_record_variation_2 = Value::Record(vec![(
            "middle_field_1".into(),
            Value::Union(1, Box::new(inner_record.clone())),
        )]);
        let outer_record_variation_1 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(0, Box::new(Value::Null)),
            ),
            ("outer_field_2".into(), inner_record.clone()),
        ]);
        let outer_record_variation_2 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(1, Box::new(middle_record_variation_1)),
            ),
            ("outer_field_2".into(), inner_record.clone()),
        ]);
        let outer_record_variation_3 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(1, Box::new(middle_record_variation_2)),
            ),
            ("outer_field_2".into(), inner_record),
        ]);

        outer_record_variation_1
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");
        outer_record_variation_2
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");
        outer_record_variation_3
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");

        Ok(())
    }

    #[test]
    fn test_avro_3461_test_multi_level_resolve_middle_namespace() -> TestResult {
        let schema = r#"
        {
          "name": "record_name",
          "namespace": "space",
          "type": "record",
          "fields": [
            {
              "name": "outer_field_1",
              "type": [
                        "null",
                        {
                            "type": "record",
                            "name": "middle_record_name",
                            "namespace":"middle_namespace",
                            "fields":[
                                {
                                    "name":"middle_field_1",
                                    "type":[
                                        "null",
                                        {
                                            "type":"record",
                                            "name":"inner_record_name",
                                            "fields":[
                                                {
                                                    "name":"inner_field_1",
                                                    "type":"double"
                                                }
                                            ]
                                        }
                                    ]
                                }
                            ]
                        }
                    ]
            },
            {
                "name": "outer_field_2",
                "type" : "middle_namespace.inner_record_name"
            }
          ]
        }
        "#;
        let schema = Schema::parse_str(schema)?;
        let inner_record = Value::Record(vec![("inner_field_1".into(), Value::Double(5.4))]);
        let middle_record_variation_1 = Value::Record(vec![(
            "middle_field_1".into(),
            Value::Union(0, Box::new(Value::Null)),
        )]);
        let middle_record_variation_2 = Value::Record(vec![(
            "middle_field_1".into(),
            Value::Union(1, Box::new(inner_record.clone())),
        )]);
        let outer_record_variation_1 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(0, Box::new(Value::Null)),
            ),
            ("outer_field_2".into(), inner_record.clone()),
        ]);
        let outer_record_variation_2 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(1, Box::new(middle_record_variation_1)),
            ),
            ("outer_field_2".into(), inner_record.clone()),
        ]);
        let outer_record_variation_3 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(1, Box::new(middle_record_variation_2)),
            ),
            ("outer_field_2".into(), inner_record),
        ]);

        outer_record_variation_1
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");
        outer_record_variation_2
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");
        outer_record_variation_3
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");

        Ok(())
    }

    #[test]
    fn test_avro_3461_test_multi_level_resolve_inner_namespace() -> TestResult {
        let schema = r#"
        {
          "name": "record_name",
          "namespace": "space",
          "type": "record",
          "fields": [
            {
              "name": "outer_field_1",
              "type": [
                        "null",
                        {
                            "type": "record",
                            "name": "middle_record_name",
                            "namespace":"middle_namespace",
                            "fields":[
                                {
                                    "name":"middle_field_1",
                                    "type":[
                                        "null",
                                        {
                                            "type":"record",
                                            "name":"inner_record_name",
                                            "namespace":"inner_namespace",
                                            "fields":[
                                                {
                                                    "name":"inner_field_1",
                                                    "type":"double"
                                                }
                                            ]
                                        }
                                    ]
                                }
                            ]
                        }
                    ]
            },
            {
                "name": "outer_field_2",
                "type" : "inner_namespace.inner_record_name"
            }
          ]
        }
        "#;
        let schema = Schema::parse_str(schema)?;

        let inner_record = Value::Record(vec![("inner_field_1".into(), Value::Double(5.4))]);
        let middle_record_variation_1 = Value::Record(vec![(
            "middle_field_1".into(),
            Value::Union(0, Box::new(Value::Null)),
        )]);
        let middle_record_variation_2 = Value::Record(vec![(
            "middle_field_1".into(),
            Value::Union(1, Box::new(inner_record.clone())),
        )]);
        let outer_record_variation_1 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(0, Box::new(Value::Null)),
            ),
            ("outer_field_2".into(), inner_record.clone()),
        ]);
        let outer_record_variation_2 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(1, Box::new(middle_record_variation_1)),
            ),
            ("outer_field_2".into(), inner_record.clone()),
        ]);
        let outer_record_variation_3 = Value::Record(vec![
            (
                "outer_field_1".into(),
                Value::Union(1, Box::new(middle_record_variation_2)),
            ),
            ("outer_field_2".into(), inner_record),
        ]);

        outer_record_variation_1
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");
        outer_record_variation_2
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");
        outer_record_variation_3
            .resolve(&schema)
            .expect("Should be able to resolve value to the schema that is it's definition");

        Ok(())
    }

    #[test]
    fn test_avro_3460_validation_with_refs() -> TestResult {
        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":{
                        "type":"record",
                        "name": "Inner",
                        "fields": [ {
                            "name":"z",
                            "type":"int"
                        }]
                    }
                },
                {
                    "name":"b",
                    "type":"Inner"
                }
            ]
        }"#,
        )?;

        let inner_value_right = Value::Record(vec![("z".into(), Value::Int(3))]);
        let inner_value_wrong1 = Value::Record(vec![("z".into(), Value::Null)]);
        let inner_value_wrong2 = Value::Record(vec![("a".into(), Value::String("testing".into()))]);
        let outer1 = Value::Record(vec![
            ("a".into(), inner_value_right.clone()),
            ("b".into(), inner_value_wrong1),
        ]);

        let outer2 = Value::Record(vec![
            ("a".into(), inner_value_right),
            ("b".into(), inner_value_wrong2),
        ]);

        assert!(
            !outer1.validate(&schema),
            "field b record is invalid against the schema"
        ); // this should pass, but doesn't
        assert!(
            !outer2.validate(&schema),
            "field b record is invalid against the schema"
        ); // this should pass, but doesn't

        Ok(())
    }

    #[test]
    fn test_avro_3460_validation_with_refs_real_struct() -> TestResult {
        use crate::ser::Serializer;
        use serde::Serialize;

        #[derive(Serialize, Clone)]
        struct TestInner {
            z: i32,
        }

        #[derive(Serialize)]
        struct TestRefSchemaStruct1 {
            a: TestInner,
            b: String, // could be literally anything
        }

        #[derive(Serialize)]
        struct TestRefSchemaStruct2 {
            a: TestInner,
            b: i32, // could be literally anything
        }

        #[derive(Serialize)]
        struct TestRefSchemaStruct3 {
            a: TestInner,
            b: Option<TestInner>, // could be literally anything
        }

        let schema = Schema::parse_str(
            r#"
        {
            "type":"record",
            "name":"TestStruct",
            "fields": [
                {
                    "name":"a",
                    "type":{
                        "type":"record",
                        "name": "Inner",
                        "fields": [ {
                            "name":"z",
                            "type":"int"
                        }]
                    }
                },
                {
                    "name":"b",
                    "type":"Inner"
                }
            ]
        }"#,
        )?;

        let test_inner = TestInner { z: 3 };
        let test_outer1 = TestRefSchemaStruct1 {
            a: test_inner.clone(),
            b: "testing".into(),
        };
        let test_outer2 = TestRefSchemaStruct2 {
            a: test_inner.clone(),
            b: 24,
        };
        let test_outer3 = TestRefSchemaStruct3 {
            a: test_inner,
            b: None,
        };

        let mut ser = Serializer::default();
        let test_outer1: Value = test_outer1.serialize(&mut ser)?;
        let mut ser = Serializer::default();
        let test_outer2: Value = test_outer2.serialize(&mut ser)?;
        let mut ser = Serializer::default();
        let test_outer3: Value = test_outer3.serialize(&mut ser)?;

        assert!(
            !test_outer1.validate(&schema),
            "field b record is invalid against the schema"
        );
        assert!(
            !test_outer2.validate(&schema),
            "field b record is invalid against the schema"
        );
        assert!(
            !test_outer3.validate(&schema),
            "field b record is invalid against the schema"
        );

        Ok(())
    }

    fn avro_3674_with_or_without_namespace(with_namespace: bool) -> TestResult {
        use crate::ser::Serializer;
        use serde::Serialize;

        let schema_str = r#"{
            "type": "record",
            "name": "NamespacedMessage",
            [NAMESPACE]
            "fields": [
                {
                    "type": "record",
                    "name": "field_a",
                    "fields": [
                        {
                            "name": "enum_a",
                            "type": {
                                "type": "enum",
                                "name": "EnumType",
                                "symbols": [
                                    "SYMBOL_1",
                                    "SYMBOL_2"
                                ],
                                "default": "SYMBOL_1"
                            }
                        },
                        {
                            "name": "enum_b",
                            "type": "EnumType"
                        }
                    ]
                }
            ]
        }"#;
        let schema_str = schema_str.replace(
            "[NAMESPACE]",
            if with_namespace {
                r#""namespace": "com.domain","#
            } else {
                ""
            },
        );

        let schema = Schema::parse_str(&schema_str)?;

        #[derive(Serialize)]
        enum EnumType {
            #[serde(rename = "SYMBOL_1")]
            Symbol1,
            #[serde(rename = "SYMBOL_2")]
            Symbol2,
        }

        #[derive(Serialize)]
        struct FieldA {
            enum_a: EnumType,
            enum_b: EnumType,
        }

        #[derive(Serialize)]
        struct NamespacedMessage {
            field_a: FieldA,
        }

        let msg = NamespacedMessage {
            field_a: FieldA {
                enum_a: EnumType::Symbol2,
                enum_b: EnumType::Symbol1,
            },
        };

        let mut ser = Serializer::default();
        let test_value: Value = msg.serialize(&mut ser)?;
        assert!(test_value.validate(&schema), "test_value should validate");
        assert!(
            test_value.resolve(&schema).is_ok(),
            "test_value should resolve"
        );

        Ok(())
    }

    #[test]
    fn test_avro_3674_validate_no_namespace_resolution() -> TestResult {
        avro_3674_with_or_without_namespace(false)
    }

    #[test]
    fn test_avro_3674_validate_with_namespace_resolution() -> TestResult {
        avro_3674_with_or_without_namespace(true)
    }

    fn avro_3688_schema_resolution_panic(set_field_b: bool) -> TestResult {
        use crate::ser::Serializer;
        use serde::{Deserialize, Serialize};

        let schema_str = r#"{
            "type": "record",
            "name": "Message",
            "fields": [
                {
                    "name": "field_a",
                    "type": [
                        "null",
                        {
                            "name": "Inner",
                            "type": "record",
                            "fields": [
                                {
                                    "name": "inner_a",
                                    "type": "string"
                                }
                            ]
                        }
                    ],
                    "default": null
                },
                {
                    "name": "field_b",
                    "type": [
                        "null",
                        "Inner"
                    ],
                    "default": null
                }
            ]
        }"#;

        #[derive(Serialize, Deserialize)]
        struct Inner {
            inner_a: String,
        }

        #[derive(Serialize, Deserialize)]
        struct Message {
            field_a: Option<Inner>,
            field_b: Option<Inner>,
        }

        let schema = Schema::parse_str(schema_str)?;

        let msg = Message {
            field_a: Some(Inner {
                inner_a: "foo".to_string(),
            }),
            field_b: if set_field_b {
                Some(Inner {
                    inner_a: "bar".to_string(),
                })
            } else {
                None
            },
        };

        let mut ser = Serializer::default();
        let test_value: Value = msg.serialize(&mut ser)?;
        assert!(test_value.validate(&schema), "test_value should validate");
        assert!(
            test_value.resolve(&schema).is_ok(),
            "test_value should resolve"
        );

        Ok(())
    }

    #[test]
    fn test_avro_3688_field_b_not_set() -> TestResult {
        avro_3688_schema_resolution_panic(false)
    }

    #[test]
    fn test_avro_3688_field_b_set() -> TestResult {
        avro_3688_schema_resolution_panic(true)
    }

    #[test]
    fn test_avro_3764_use_resolve_schemata() -> TestResult {
        let referenced_schema =
            r#"{"name": "enumForReference", "type": "enum", "symbols": ["A", "B"]}"#;
        let main_schema = r#"{"name": "recordWithReference", "type": "record", "fields": [{"name": "reference", "type": "enumForReference"}]}"#;

        let value: serde_json::Value = serde_json::from_str(
            r#"
            {
                "reference": "A"
            }
        "#,
        )?;

        let avro_value = Value::from(value);

        let schemas = Schema::parse_list(&[main_schema, referenced_schema])?;

        let main_schema = schemas.get(0).unwrap();
        let schemata: Vec<_> = schemas.iter().skip(1).collect();

        let resolve_result = avro_value.clone().resolve_schemata(main_schema, schemata);

        assert!(
            resolve_result.is_ok(),
            "result of resolving with schemata should be ok, got: {:?}",
            resolve_result
        );

        let resolve_result = avro_value.resolve(main_schema);
        assert!(
            resolve_result.is_err(),
            "result of resolving without schemata should be err, got: {:?}",
            resolve_result
        );

        Ok(())
    }

    #[test]
    fn test_avro_3767_union_resolve_complex_refs() -> TestResult {
        let referenced_enum =
            r#"{"name": "enumForReference", "type": "enum", "symbols": ["A", "B"]}"#;
        let referenced_record = r#"{"name": "recordForReference", "type": "record", "fields": [{"name": "refInRecord", "type": "enumForReference"}]}"#;
        let main_schema = r#"{"name": "recordWithReference", "type": "record", "fields": [{"name": "reference", "type": ["null", "recordForReference"]}]}"#;

        let value: serde_json::Value = serde_json::from_str(
            r#"
            {
                "reference": {
                    "refInRecord": "A"
                }
            }
        "#,
        )?;

        let avro_value = Value::from(value);

        let schemata = Schema::parse_list(&[referenced_enum, referenced_record, main_schema])?;

        let main_schema = schemata.last().unwrap();
        let other_schemata: Vec<&Schema> = schemata.iter().take(2).collect();

        let resolve_result = avro_value.resolve_schemata(main_schema, other_schemata);

        assert!(
            resolve_result.is_ok(),
            "result of resolving with schemata should be ok, got: {:?}",
            resolve_result
        );

        assert!(
            resolve_result?.validate_schemata(schemata.iter().collect()),
            "result of validation with schemata should be true"
        );

        Ok(())
    }

    #[test]
    fn test_avro_3782_incorrect_decimal_resolving() -> TestResult {
        let schema = r#"{"name": "decimalSchema", "logicalType": "decimal", "type": "fixed", "precision": 8, "scale": 0, "size": 8}"#;

        let avro_value = Value::Decimal(Decimal::from(
            BigInt::from(12345678u32).to_signed_bytes_be(),
        ));
        let schema = Schema::parse_str(schema)?;
        let resolve_result = avro_value.resolve(&schema);
        assert!(
            resolve_result.is_ok(),
            "resolve result must be ok, got: {resolve_result:?}"
        );

        Ok(())
    }
}