This is an automated email from the ASF dual-hosted git repository. mgrigorov pushed a commit to branch avro-3814/schema-resolution-union in repository https://gitbox.apache.org/repos/asf/avro.git
commit 11249ae28333e9fc827f311b24bd9f9b4bb9e035 Author: Rik Heijdens <[email protected]> AuthorDate: Fri Jul 28 11:16:54 2023 +0200 AVRO-3814: Fix schema resolution for records in union types The logic for validation records in Value::validate_internal() would be too strict when resolving union types containing a record. This could lead to a situation where schema resolution would fail because the correct schema to use for a union type could not be identified. This commit fixes this by passing a boolean `schema_resolution` to `Value::validate_internal()` which governs whether schema_resolution rules should be applied. --- lang/rust/avro/src/schema.rs | 5 ++-- lang/rust/avro/src/types.rs | 61 +++++++++++++++++++++++++++++++++++--------- lang/rust/avro/src/writer.rs | 10 ++++++-- 3 files changed, 60 insertions(+), 16 deletions(-) diff --git a/lang/rust/avro/src/schema.rs b/lang/rust/avro/src/schema.rs index 399668641..e586d05f5 100644 --- a/lang/rust/avro/src/schema.rs +++ b/lang/rust/avro/src/schema.rs @@ -814,7 +814,6 @@ impl UnionSchema { Some((i, &self.schemas[i])) } else { // slow path (required for matching logical or named types) - // first collect what schemas we already know let mut collected_names: HashMap<Name, &Schema> = known_schemata .map(|names| { @@ -838,8 +837,10 @@ impl UnionSchema { // extend known schemas with just resolved names collected_names.extend(resolved_names); let namespace = &schema.namespace().or_else(|| enclosing_namespace.clone()); + + // Attempt to validate the value in order to ensure we've selected the right schema. value - .validate_internal(schema, &collected_names, namespace) + .validate_internal(schema, &collected_names, namespace, true) .is_none() }) } diff --git a/lang/rust/avro/src/types.rs b/lang/rust/avro/src/types.rs index fbc4fa004..28809f0ea 100644 --- a/lang/rust/avro/src/types.rs +++ b/lang/rust/avro/src/types.rs @@ -350,7 +350,7 @@ impl Value { schemata.iter().any(|schema| { let enclosing_namespace = schema.namespace(); - match self.validate_internal(schema, rs.get_names(), &enclosing_namespace) { + match self.validate_internal(schema, rs.get_names(), &enclosing_namespace, false) { Some(reason) => { let log_message = format!( "Invalid value: {:?} for schema: {:?}. Reason: {}", @@ -377,11 +377,16 @@ impl Value { } } + /// Validates the value against the provided schema. + /// + /// Arguments: + /// * `schema_resolution` - whether schema resolution rules should be applied when validating the `value`. pub(crate) fn validate_internal<S: std::borrow::Borrow<Schema> + Debug>( &self, schema: &Schema, names: &HashMap<Name, S>, enclosing_namespace: &Namespace, + schema_resolution: bool, ) -> Option<String> { match (self, schema) { (_, Schema::Ref { name }) => { @@ -394,7 +399,14 @@ impl Value { names.keys() )) }, - |s| self.validate_internal(s.borrow(), names, &name.namespace), + |s| { + self.validate_internal( + s.borrow(), + names, + &name.namespace, + schema_resolution, + ) + }, ) } (&Value::Null, &Schema::Null) => None, @@ -482,7 +494,9 @@ impl Value { (&Value::Union(i, ref value), Schema::Union(inner)) => inner .variants() .get(i as usize) - .map(|schema| value.validate_internal(schema, names, enclosing_namespace)) + .map(|schema| { + value.validate_internal(schema, names, enclosing_namespace, schema_resolution) + }) .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) { @@ -493,14 +507,19 @@ impl Value { (Value::Array(items), Schema::Array(inner)) => items.iter().fold(None, |acc, item| { Value::accumulate( acc, - item.validate_internal(inner, names, enclosing_namespace), + item.validate_internal(inner, names, enclosing_namespace, schema_resolution), ) }), (Value::Map(items), Schema::Map(inner)) => { items.iter().fold(None, |acc, (_, value)| { Value::accumulate( acc, - value.validate_internal(inner, names, enclosing_namespace), + value.validate_internal( + inner, + names, + enclosing_namespace, + schema_resolution, + ), ) }) } @@ -516,13 +535,14 @@ impl Value { 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() { + } else if record_fields.len() > fields.len() && !schema_resolution { return Some(format!( "The value's records length ({}) is greater than the schema's ({} fields)", record_fields.len(), @@ -547,20 +567,37 @@ impl Value { &field.schema, names, record_namespace, + schema_resolution, ), ) } - None => Value::accumulate( - acc, - Some(format!("There is no schema field for field '{field_name}'")), - ), + None => { + if schema_resolution { + // While performing validation during schema resolution we allow + // extraneous fields to exist in the Value::Record, as these + // will get cleaned up later by the schema resolution logic. + acc + } else { + 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); + let res = item.validate_internal( + &field.schema, + names, + enclosing_namespace, + schema_resolution, + ); Value::accumulate(acc, res) } else if !field.is_nullable() { Value::accumulate( @@ -1252,7 +1289,7 @@ mod tests { for (value, schema, valid, expected_err_message) in value_schema_valid.into_iter() { let err_message = - value.validate_internal::<Schema>(&schema, &HashMap::default(), &None); + value.validate_internal::<Schema>(&schema, &HashMap::default(), &None, false); assert_eq!(valid, err_message.is_none()); if !valid { let full_err_message = format!( diff --git a/lang/rust/avro/src/writer.rs b/lang/rust/avro/src/writer.rs index 83e863455..79641ed0c 100644 --- a/lang/rust/avro/src/writer.rs +++ b/lang/rust/avro/src/writer.rs @@ -425,7 +425,7 @@ fn write_avro_datum_schemata<T: Into<Value>>( let rs = ResolvedSchema::try_from(schemata)?; let names = rs.get_names(); let enclosing_namespace = schema.namespace(); - if let Some(_err) = avro.validate_internal(schema, names, &enclosing_namespace) { + if let Some(_err) = avro.validate_internal(schema, names, &enclosing_namespace, false) { return Err(Error::Validation); } encode_internal(&avro, schema, names, &enclosing_namespace, buffer) @@ -544,7 +544,12 @@ fn write_value_ref_resolved( value: &Value, buffer: &mut Vec<u8>, ) -> AvroResult<()> { - match value.validate_internal(schema, resolved_schema.get_names(), &schema.namespace()) { + match value.validate_internal( + schema, + resolved_schema.get_names(), + &schema.namespace(), + false, + ) { Some(err) => Err(Error::ValidationWithReason(err)), None => encode_internal( value, @@ -566,6 +571,7 @@ fn write_value_ref_owned_resolved( root_schema, resolved_schema.get_names(), &root_schema.namespace(), + false, ) { return Err(Error::ValidationWithReason(err)); }
