fn match_record_schemas()

in avro/src/schema_compatibility.rs [172:240]

• 61 lines of code
• 11 McCabe index (conditional complexity)

    fn match_record_schemas(
        &mut self,
        writers_schema: &Schema,
        readers_schema: &Schema,
    ) -> Result<(), CompatibilityError> {
        let w_type = SchemaKind::from(writers_schema);

        if w_type == SchemaKind::Union {
            return Err(CompatibilityError::TypeExpected {
                schema_type: String::from("writers_schema"),
                expected_type: vec![SchemaKind::Record],
            });
        }

        if let Schema::Record(RecordSchema {
            fields: w_fields,
            lookup: w_lookup,
            ..
        }) = writers_schema
        {
            if let Schema::Record(RecordSchema {
                fields: r_fields, ..
            }) = readers_schema
            {
                for field in r_fields.iter() {
                    // get all field names in a vector (field.name + aliases)
                    let mut fields_names = vec![&field.name];
                    if let Some(ref aliases) = field.aliases {
                        for alias in aliases {
                            fields_names.push(alias);
                        }
                    }

                    // Check whether any of the possible fields names are in the writer schema.
                    // If the field was found, then it must have the exact same name with the writer field,
                    // otherwise we would have a false positive with the writers aliases
                    let position = fields_names.iter().find_map(|field_name| {
                        if let Some(pos) = w_lookup.get(*field_name) {
                            if &w_fields[*pos].name == *field_name {
                                return Some(pos);
                            }
                        }
                        None
                    });

                    match position {
                        Some(pos) => {
                            if let Err(err) =
                                self.full_match_schemas(&w_fields[*pos].schema, &field.schema)
                            {
                                return Err(CompatibilityError::FieldTypeMismatch(
                                    field.name.clone(),
                                    Box::new(err),
                                ));
                            }
                        }
                        _ => {
                            if field.default.is_none() {
                                return Err(CompatibilityError::MissingDefaultValue(
                                    field.name.clone(),
                                ));
                            }
                        }
                    }
                }
            }
        }
        Ok(())
    }