| // 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 for checking schema compatibility |
| use crate::schema::{Schema, SchemaKind}; |
| use std::{ |
| collections::{hash_map::DefaultHasher, HashSet}, |
| hash::Hasher, |
| ptr, |
| }; |
| |
| pub struct SchemaCompatibility; |
| |
| struct Checker { |
| recursion: HashSet<(u64, u64)>, |
| } |
| |
| impl Checker { |
| /// Create a new checker, with recursion set to an empty set. |
| pub(crate) fn new() -> Self { |
| Self { |
| recursion: HashSet::new(), |
| } |
| } |
| |
| pub(crate) fn can_read(&mut self, writers_schema: &Schema, readers_schema: &Schema) -> bool { |
| self.full_match_schemas(writers_schema, readers_schema) |
| } |
| |
| pub(crate) fn full_match_schemas( |
| &mut self, |
| writers_schema: &Schema, |
| readers_schema: &Schema, |
| ) -> bool { |
| if self.recursion_in_progress(writers_schema, readers_schema) { |
| return true; |
| } |
| |
| if !SchemaCompatibility::match_schemas(writers_schema, readers_schema) { |
| return false; |
| } |
| |
| let w_type = SchemaKind::from(writers_schema); |
| let r_type = SchemaKind::from(readers_schema); |
| |
| if w_type != SchemaKind::Union && (r_type.is_primitive() || r_type == SchemaKind::Fixed) { |
| return true; |
| } |
| |
| match r_type { |
| SchemaKind::Record => self.match_record_schemas(writers_schema, readers_schema), |
| SchemaKind::Map => { |
| if let Schema::Map(w_m) = writers_schema { |
| if let Schema::Map(r_m) = readers_schema { |
| self.full_match_schemas(w_m, r_m) |
| } else { |
| unreachable!("readers_schema should have been Schema::Map") |
| } |
| } else { |
| unreachable!("writers_schema should have been Schema::Map") |
| } |
| } |
| SchemaKind::Array => { |
| if let Schema::Array(w_a) = writers_schema { |
| if let Schema::Array(r_a) = readers_schema { |
| self.full_match_schemas(w_a, r_a) |
| } else { |
| unreachable!("readers_schema should have been Schema::Array") |
| } |
| } else { |
| unreachable!("writers_schema should have been Schema::Array") |
| } |
| } |
| SchemaKind::Union => self.match_union_schemas(writers_schema, readers_schema), |
| SchemaKind::Enum => { |
| // reader's symbols must contain all writer's symbols |
| if let Schema::Enum { |
| symbols: w_symbols, .. |
| } = writers_schema |
| { |
| if let Schema::Enum { |
| symbols: r_symbols, .. |
| } = readers_schema |
| { |
| return !w_symbols.iter().any(|e| !r_symbols.contains(e)); |
| } |
| } |
| false |
| } |
| _ => { |
| if w_type == SchemaKind::Union { |
| if let Schema::Union(r) = writers_schema { |
| if r.schemas.len() == 1 { |
| return self.full_match_schemas(&r.schemas[0], readers_schema); |
| } |
| } |
| } |
| false |
| } |
| } |
| } |
| |
| fn match_record_schemas(&mut self, writers_schema: &Schema, readers_schema: &Schema) -> bool { |
| let w_type = SchemaKind::from(writers_schema); |
| |
| if w_type == SchemaKind::Union { |
| return false; |
| } |
| |
| if let Schema::Record { |
| fields: w_fields, |
| lookup: w_lookup, |
| .. |
| } = writers_schema |
| { |
| if let Schema::Record { |
| fields: r_fields, .. |
| } = readers_schema |
| { |
| for field in r_fields.iter() { |
| if let Some(pos) = w_lookup.get(&field.name) { |
| if !self.full_match_schemas(&w_fields[*pos].schema, &field.schema) { |
| return false; |
| } |
| } else if field.default.is_none() { |
| return false; |
| } |
| } |
| } |
| } |
| true |
| } |
| |
| fn match_union_schemas(&mut self, writers_schema: &Schema, readers_schema: &Schema) -> bool { |
| let w_type = SchemaKind::from(writers_schema); |
| let r_type = SchemaKind::from(readers_schema); |
| |
| assert_eq!(r_type, SchemaKind::Union); |
| |
| if w_type == SchemaKind::Union { |
| if let Schema::Union(u) = writers_schema { |
| u.schemas |
| .iter() |
| .all(|schema| self.full_match_schemas(schema, readers_schema)) |
| } else { |
| unreachable!("writers_schema should have been Schema::Union") |
| } |
| } else if let Schema::Union(u) = readers_schema { |
| u.schemas |
| .iter() |
| .any(|schema| self.full_match_schemas(writers_schema, schema)) |
| } else { |
| unreachable!("readers_schema should have been Schema::Union") |
| } |
| } |
| |
| fn recursion_in_progress(&mut self, writers_schema: &Schema, readers_schema: &Schema) -> bool { |
| let mut hasher = DefaultHasher::new(); |
| ptr::hash(writers_schema, &mut hasher); |
| let w_hash = hasher.finish(); |
| |
| hasher = DefaultHasher::new(); |
| ptr::hash(readers_schema, &mut hasher); |
| let r_hash = hasher.finish(); |
| |
| let key = (w_hash, r_hash); |
| // This is a shortcut to add if not exists *and* return false. It will return true |
| // if it was able to insert. |
| !self.recursion.insert(key) |
| } |
| } |
| |
| impl SchemaCompatibility { |
| /// `can_read` performs a full, recursive check that a datum written using the |
| /// writers_schema can be read using the readers_schema. |
| pub fn can_read(writers_schema: &Schema, readers_schema: &Schema) -> bool { |
| let mut c = Checker::new(); |
| c.can_read(writers_schema, readers_schema) |
| } |
| |
| /// `mutual_read` performs a full, recursive check that a datum written using either |
| /// the writers_schema or the readers_schema can be read using the other schema. |
| pub fn mutual_read(writers_schema: &Schema, readers_schema: &Schema) -> bool { |
| SchemaCompatibility::can_read(writers_schema, readers_schema) |
| && SchemaCompatibility::can_read(readers_schema, writers_schema) |
| } |
| |
| /// `match_schemas` performs a basic check that a datum written with the |
| /// writers_schema could be read using the readers_schema. This check only includes |
| /// matching the types, including schema promotion, and matching the full name for |
| /// named types. Aliases for named types are not supported here, and the rust |
| /// implementation of Avro in general does not include support for aliases (I think). |
| pub(crate) fn match_schemas(writers_schema: &Schema, readers_schema: &Schema) -> bool { |
| let w_type = SchemaKind::from(writers_schema); |
| let r_type = SchemaKind::from(readers_schema); |
| |
| if w_type == SchemaKind::Union || r_type == SchemaKind::Union { |
| return true; |
| } |
| |
| if w_type == r_type { |
| if r_type.is_primitive() { |
| return true; |
| } |
| |
| match r_type { |
| SchemaKind::Record => { |
| if let Schema::Record { name: w_name, .. } = writers_schema { |
| if let Schema::Record { name: r_name, .. } = readers_schema { |
| return w_name.fullname(None) == r_name.fullname(None); |
| } else { |
| unreachable!("readers_schema should have been Schema::Record") |
| } |
| } else { |
| unreachable!("writers_schema should have been Schema::Record") |
| } |
| } |
| SchemaKind::Fixed => { |
| if let Schema::Fixed { |
| name: w_name, |
| aliases: _, |
| doc: _w_doc, |
| size: w_size, |
| } = writers_schema |
| { |
| if let Schema::Fixed { |
| name: r_name, |
| aliases: _, |
| doc: _r_doc, |
| size: r_size, |
| } = readers_schema |
| { |
| return w_name.fullname(None) == r_name.fullname(None) |
| && w_size == r_size; |
| } else { |
| unreachable!("readers_schema should have been Schema::Fixed") |
| } |
| } else { |
| unreachable!("writers_schema should have been Schema::Fixed") |
| } |
| } |
| SchemaKind::Enum => { |
| if let Schema::Enum { name: w_name, .. } = writers_schema { |
| if let Schema::Enum { name: r_name, .. } = readers_schema { |
| return w_name.fullname(None) == r_name.fullname(None); |
| } else { |
| unreachable!("readers_schema should have been Schema::Enum") |
| } |
| } else { |
| unreachable!("writers_schema should have been Schema::Enum") |
| } |
| } |
| SchemaKind::Map => { |
| if let Schema::Map(w_m) = writers_schema { |
| if let Schema::Map(r_m) = readers_schema { |
| return SchemaCompatibility::match_schemas(w_m, r_m); |
| } else { |
| unreachable!("readers_schema should have been Schema::Map") |
| } |
| } else { |
| unreachable!("writers_schema should have been Schema::Map") |
| } |
| } |
| SchemaKind::Array => { |
| if let Schema::Array(w_a) = writers_schema { |
| if let Schema::Array(r_a) = readers_schema { |
| return SchemaCompatibility::match_schemas(w_a, r_a); |
| } else { |
| unreachable!("readers_schema should have been Schema::Array") |
| } |
| } else { |
| unreachable!("writers_schema should have been Schema::Array") |
| } |
| } |
| _ => (), |
| }; |
| } |
| |
| if w_type == SchemaKind::Int |
| && vec![SchemaKind::Long, SchemaKind::Float, SchemaKind::Double] |
| .iter() |
| .any(|&t| t == r_type) |
| { |
| return true; |
| } |
| |
| if w_type == SchemaKind::Long |
| && vec![SchemaKind::Float, SchemaKind::Double] |
| .iter() |
| .any(|&t| t == r_type) |
| { |
| return true; |
| } |
| |
| if w_type == SchemaKind::Float && r_type == SchemaKind::Double { |
| return true; |
| } |
| |
| if w_type == SchemaKind::String && r_type == SchemaKind::Bytes { |
| return true; |
| } |
| |
| if w_type == SchemaKind::Bytes && r_type == SchemaKind::String { |
| return true; |
| } |
| |
| false |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| |
| fn int_array_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"array", "items":"int"}"#).unwrap() |
| } |
| |
| fn long_array_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"array", "items":"long"}"#).unwrap() |
| } |
| |
| fn string_array_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"array", "items":"string"}"#).unwrap() |
| } |
| |
| fn int_map_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"map", "values":"int"}"#).unwrap() |
| } |
| |
| fn long_map_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"map", "values":"long"}"#).unwrap() |
| } |
| |
| fn string_map_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"map", "values":"string"}"#).unwrap() |
| } |
| |
| fn enum1_ab_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"enum", "name":"Enum1", "symbols":["A","B"]}"#).unwrap() |
| } |
| |
| fn enum1_abc_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"enum", "name":"Enum1", "symbols":["A","B","C"]}"#).unwrap() |
| } |
| |
| fn enum1_bc_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"enum", "name":"Enum1", "symbols":["B","C"]}"#).unwrap() |
| } |
| |
| fn enum2_ab_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"enum", "name":"Enum2", "symbols":["A","B"]}"#).unwrap() |
| } |
| |
| fn empty_record1_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"record", "name":"Record1", "fields":[]}"#).unwrap() |
| } |
| |
| fn empty_record2_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"record", "name":"Record2", "fields": []}"#).unwrap() |
| } |
| |
| fn a_int_record1_schema() -> Schema { |
| Schema::parse_str( |
| r#"{"type":"record", "name":"Record1", "fields":[{"name":"a", "type":"int"}]}"#, |
| ) |
| .unwrap() |
| } |
| |
| fn a_long_record1_schema() -> Schema { |
| Schema::parse_str( |
| r#"{"type":"record", "name":"Record1", "fields":[{"name":"a", "type":"long"}]}"#, |
| ) |
| .unwrap() |
| } |
| |
| fn a_int_b_int_record1_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"record", "name":"Record1", "fields":[{"name":"a", "type":"int"}, {"name":"b", "type":"int"}]}"#).unwrap() |
| } |
| |
| fn a_dint_record1_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"record", "name":"Record1", "fields":[{"name":"a", "type":"int", "default":0}]}"#).unwrap() |
| } |
| |
| fn a_int_b_dint_record1_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"record", "name":"Record1", "fields":[{"name":"a", "type":"int"}, {"name":"b", "type":"int", "default":0}]}"#).unwrap() |
| } |
| |
| fn a_dint_b_dint_record1_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"record", "name":"Record1", "fields":[{"name":"a", "type":"int", "default":0}, {"name":"b", "type":"int", "default":0}]}"#).unwrap() |
| } |
| |
| fn nested_record() -> Schema { |
| Schema::parse_str(r#"{"type":"record","name":"parent","fields":[{"name":"attribute","type":{"type":"record","name":"child","fields":[{"name":"id","type":"string"}]}}]}"#).unwrap() |
| } |
| |
| fn nested_optional_record() -> Schema { |
| Schema::parse_str(r#"{"type":"record","name":"parent","fields":[{"name":"attribute","type":["null",{"type":"record","name":"child","fields":[{"name":"id","type":"string"}]}],"default":null}]}"#).unwrap() |
| } |
| |
| fn int_list_record_schema() -> Schema { |
| Schema::parse_str(r#"{"type":"record", "name":"List", "fields": [{"name": "head", "type": "int"},{"name": "tail", "type": "array", "items": "int"}]}"#).unwrap() |
| } |
| |
| fn long_list_record_schema() -> Schema { |
| Schema::parse_str( |
| r#" |
| { |
| "type":"record", "name":"List", "fields": [ |
| {"name": "head", "type": "long"}, |
| {"name": "tail", "type": "array", "items": "long"} |
| ]} |
| "#, |
| ) |
| .unwrap() |
| } |
| |
| fn union_schema(schemas: Vec<Schema>) -> Schema { |
| let schema_string = schemas |
| .iter() |
| .map(|s| s.canonical_form()) |
| .collect::<Vec<String>>() |
| .join(","); |
| Schema::parse_str(&format!("[{}]", schema_string)).unwrap() |
| } |
| |
| fn empty_union_schema() -> Schema { |
| union_schema(vec![]) |
| } |
| |
| // unused |
| // fn null_union_schema() -> Schema { union_schema(vec![Schema::Null]) } |
| |
| fn int_union_schema() -> Schema { |
| union_schema(vec![Schema::Int]) |
| } |
| |
| fn long_union_schema() -> Schema { |
| union_schema(vec![Schema::Long]) |
| } |
| |
| fn string_union_schema() -> Schema { |
| union_schema(vec![Schema::String]) |
| } |
| |
| fn int_string_union_schema() -> Schema { |
| union_schema(vec![Schema::Int, Schema::String]) |
| } |
| |
| fn string_int_union_schema() -> Schema { |
| union_schema(vec![Schema::String, Schema::Int]) |
| } |
| |
| #[test] |
| fn test_broken() { |
| assert!(!SchemaCompatibility::can_read( |
| &int_string_union_schema(), |
| &int_union_schema() |
| )) |
| } |
| |
| #[test] |
| fn test_incompatible_reader_writer_pairs() { |
| let incompatible_schemas = vec![ |
| // null |
| (Schema::Null, Schema::Int), |
| (Schema::Null, Schema::Long), |
| // boolean |
| (Schema::Boolean, Schema::Int), |
| // int |
| (Schema::Int, Schema::Null), |
| (Schema::Int, Schema::Boolean), |
| (Schema::Int, Schema::Long), |
| (Schema::Int, Schema::Float), |
| (Schema::Int, Schema::Double), |
| // long |
| (Schema::Long, Schema::Float), |
| (Schema::Long, Schema::Double), |
| // float |
| (Schema::Float, Schema::Double), |
| // string |
| (Schema::String, Schema::Boolean), |
| (Schema::String, Schema::Int), |
| // bytes |
| (Schema::Bytes, Schema::Null), |
| (Schema::Bytes, Schema::Int), |
| // array and maps |
| (int_array_schema(), long_array_schema()), |
| (int_map_schema(), int_array_schema()), |
| (int_array_schema(), int_map_schema()), |
| (int_map_schema(), long_map_schema()), |
| // enum |
| (enum1_ab_schema(), enum1_abc_schema()), |
| (enum1_bc_schema(), enum1_abc_schema()), |
| (enum1_ab_schema(), enum2_ab_schema()), |
| (Schema::Int, enum2_ab_schema()), |
| (enum2_ab_schema(), Schema::Int), |
| //union |
| (int_union_schema(), int_string_union_schema()), |
| (string_union_schema(), int_string_union_schema()), |
| //record |
| (empty_record2_schema(), empty_record1_schema()), |
| (a_int_record1_schema(), empty_record1_schema()), |
| (a_int_b_dint_record1_schema(), empty_record1_schema()), |
| (int_list_record_schema(), long_list_record_schema()), |
| (nested_record(), nested_optional_record()), |
| ]; |
| |
| assert!(!incompatible_schemas |
| .iter() |
| .any(|(reader, writer)| SchemaCompatibility::can_read(writer, reader))); |
| } |
| |
| #[test] |
| fn test_compatible_reader_writer_pairs() { |
| let compatible_schemas = vec![ |
| (Schema::Null, Schema::Null), |
| (Schema::Long, Schema::Int), |
| (Schema::Float, Schema::Int), |
| (Schema::Float, Schema::Long), |
| (Schema::Double, Schema::Long), |
| (Schema::Double, Schema::Int), |
| (Schema::Double, Schema::Float), |
| (Schema::String, Schema::Bytes), |
| (Schema::Bytes, Schema::String), |
| (int_array_schema(), int_array_schema()), |
| (long_array_schema(), int_array_schema()), |
| (int_map_schema(), int_map_schema()), |
| (long_map_schema(), int_map_schema()), |
| (enum1_ab_schema(), enum1_ab_schema()), |
| (enum1_abc_schema(), enum1_ab_schema()), |
| (empty_union_schema(), empty_union_schema()), |
| (int_union_schema(), int_union_schema()), |
| (int_string_union_schema(), string_int_union_schema()), |
| (int_union_schema(), empty_union_schema()), |
| (long_union_schema(), int_union_schema()), |
| (int_union_schema(), Schema::Int), |
| (Schema::Int, int_union_schema()), |
| (empty_record1_schema(), empty_record1_schema()), |
| (empty_record1_schema(), a_int_record1_schema()), |
| (a_int_record1_schema(), a_int_record1_schema()), |
| (a_dint_record1_schema(), a_int_record1_schema()), |
| (a_dint_record1_schema(), a_dint_record1_schema()), |
| (a_int_record1_schema(), a_dint_record1_schema()), |
| (a_long_record1_schema(), a_int_record1_schema()), |
| (a_int_record1_schema(), a_int_b_int_record1_schema()), |
| (a_dint_record1_schema(), a_int_b_int_record1_schema()), |
| (a_int_b_dint_record1_schema(), a_int_record1_schema()), |
| (a_dint_b_dint_record1_schema(), empty_record1_schema()), |
| (a_dint_b_dint_record1_schema(), a_int_record1_schema()), |
| (a_int_b_int_record1_schema(), a_dint_b_dint_record1_schema()), |
| (int_list_record_schema(), int_list_record_schema()), |
| (long_list_record_schema(), long_list_record_schema()), |
| (long_list_record_schema(), int_list_record_schema()), |
| (nested_optional_record(), nested_record()), |
| ]; |
| |
| assert!(compatible_schemas |
| .iter() |
| .all(|(reader, writer)| SchemaCompatibility::can_read(writer, reader))); |
| } |
| |
| fn writer_schema() -> Schema { |
| Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Record", "fields":[ |
| {"name":"oldfield1", "type":"int"}, |
| {"name":"oldfield2", "type":"string"} |
| ]} |
| "#, |
| ) |
| .unwrap() |
| } |
| |
| #[test] |
| fn test_missing_field() { |
| let reader_schema = Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Record", "fields":[ |
| {"name":"oldfield1", "type":"int"} |
| ]} |
| "#, |
| ) |
| .unwrap(); |
| assert!(SchemaCompatibility::can_read( |
| &writer_schema(), |
| &reader_schema, |
| )); |
| assert!(!SchemaCompatibility::can_read( |
| &reader_schema, |
| &writer_schema() |
| )); |
| } |
| |
| #[test] |
| fn test_missing_second_field() { |
| let reader_schema = Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Record", "fields":[ |
| {"name":"oldfield2", "type":"string"} |
| ]} |
| "#, |
| ) |
| .unwrap(); |
| assert!(SchemaCompatibility::can_read( |
| &writer_schema(), |
| &reader_schema |
| )); |
| assert!(!SchemaCompatibility::can_read( |
| &reader_schema, |
| &writer_schema() |
| )); |
| } |
| |
| #[test] |
| fn test_all_fields() { |
| let reader_schema = Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Record", "fields":[ |
| {"name":"oldfield1", "type":"int"}, |
| {"name":"oldfield2", "type":"string"} |
| ]} |
| "#, |
| ) |
| .unwrap(); |
| assert!(SchemaCompatibility::can_read( |
| &writer_schema(), |
| &reader_schema |
| )); |
| assert!(SchemaCompatibility::can_read( |
| &reader_schema, |
| &writer_schema() |
| )); |
| } |
| |
| #[test] |
| fn test_new_field_with_default() { |
| let reader_schema = Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Record", "fields":[ |
| {"name":"oldfield1", "type":"int"}, |
| {"name":"newfield1", "type":"int", "default":42} |
| ]} |
| "#, |
| ) |
| .unwrap(); |
| assert!(SchemaCompatibility::can_read( |
| &writer_schema(), |
| &reader_schema |
| )); |
| assert!(!SchemaCompatibility::can_read( |
| &reader_schema, |
| &writer_schema() |
| )); |
| } |
| |
| #[test] |
| fn test_new_field() { |
| let reader_schema = Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Record", "fields":[ |
| {"name":"oldfield1", "type":"int"}, |
| {"name":"newfield1", "type":"int"} |
| ]} |
| "#, |
| ) |
| .unwrap(); |
| assert!(!SchemaCompatibility::can_read( |
| &writer_schema(), |
| &reader_schema |
| )); |
| assert!(!SchemaCompatibility::can_read( |
| &reader_schema, |
| &writer_schema() |
| )); |
| } |
| |
| #[test] |
| fn test_array_writer_schema() { |
| let valid_reader = string_array_schema(); |
| let invalid_reader = string_map_schema(); |
| |
| assert!(SchemaCompatibility::can_read( |
| &string_array_schema(), |
| &valid_reader |
| )); |
| assert!(!SchemaCompatibility::can_read( |
| &string_array_schema(), |
| &invalid_reader |
| )); |
| } |
| |
| #[test] |
| fn test_primitive_writer_schema() { |
| let valid_reader = Schema::String; |
| assert!(SchemaCompatibility::can_read( |
| &Schema::String, |
| &valid_reader |
| )); |
| assert!(!SchemaCompatibility::can_read( |
| &Schema::Int, |
| &Schema::String |
| )); |
| } |
| |
| #[test] |
| fn test_union_reader_writer_subset_incompatiblity() { |
| // reader union schema must contain all writer union branches |
| let union_writer = union_schema(vec![Schema::Int, Schema::String]); |
| let union_reader = union_schema(vec![Schema::String]); |
| |
| assert!(!SchemaCompatibility::can_read(&union_writer, &union_reader)); |
| assert!(SchemaCompatibility::can_read(&union_reader, &union_writer)); |
| } |
| |
| #[test] |
| fn test_incompatible_record_field() { |
| let string_schema = Schema::parse_str( |
| r#" |
| {"type":"record", "name":"MyRecord", "namespace":"ns", "fields": [ |
| {"name":"field1", "type":"string"} |
| ]} |
| "#, |
| ) |
| .unwrap(); |
| |
| let int_schema = Schema::parse_str( |
| r#" |
| {"type":"record", "name":"MyRecord", "namespace":"ns", "fields": [ |
| {"name":"field1", "type":"int"} |
| ]} |
| "#, |
| ) |
| .unwrap(); |
| |
| assert!(!SchemaCompatibility::can_read(&string_schema, &int_schema)); |
| } |
| |
| #[test] |
| fn test_enum_symbols() { |
| let enum_schema1 = Schema::parse_str( |
| r#" |
| {"type":"enum", "name":"MyEnum", "symbols":["A","B"]} |
| "#, |
| ) |
| .unwrap(); |
| let enum_schema2 = |
| Schema::parse_str(r#"{"type":"enum", "name":"MyEnum", "symbols":["A","B","C"]}"#) |
| .unwrap(); |
| assert!(!SchemaCompatibility::can_read(&enum_schema2, &enum_schema1)); |
| assert!(SchemaCompatibility::can_read(&enum_schema1, &enum_schema2)); |
| } |
| |
| fn point_2d_schema() -> Schema { |
| Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Point2D", "fields":[ |
| {"name":"x", "type":"double"}, |
| {"name":"y", "type":"double"} |
| ]} |
| "#, |
| ) |
| .unwrap() |
| } |
| |
| fn point_2d_fullname_schema() -> Schema { |
| Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Point", "namespace":"written", "fields":[ |
| {"name":"x", "type":"double"}, |
| {"name":"y", "type":"double"} |
| ]} |
| "#, |
| ) |
| .unwrap() |
| } |
| |
| fn point_3d_no_default_schema() -> Schema { |
| Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Point", "fields":[ |
| {"name":"x", "type":"double"}, |
| {"name":"y", "type":"double"}, |
| {"name":"z", "type":"double"} |
| ]} |
| "#, |
| ) |
| .unwrap() |
| } |
| |
| fn point_3d_schema() -> Schema { |
| Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Point3D", "fields":[ |
| {"name":"x", "type":"double"}, |
| {"name":"y", "type":"double"}, |
| {"name":"z", "type":"double", "default": 0.0} |
| ]} |
| "#, |
| ) |
| .unwrap() |
| } |
| |
| fn point_3d_match_name_schema() -> Schema { |
| Schema::parse_str( |
| r#" |
| {"type":"record", "name":"Point", "fields":[ |
| {"name":"x", "type":"double"}, |
| {"name":"y", "type":"double"}, |
| {"name":"z", "type":"double", "default": 0.0} |
| ]} |
| "#, |
| ) |
| .unwrap() |
| } |
| |
| #[test] |
| fn test_union_resolution_no_structure_match() { |
| // short name match, but no structure match |
| let read_schema = union_schema(vec![Schema::Null, point_3d_no_default_schema()]); |
| assert!(!SchemaCompatibility::can_read( |
| &point_2d_fullname_schema(), |
| &read_schema |
| )); |
| } |
| |
| #[test] |
| fn test_union_resolution_first_structure_match_2d() { |
| // multiple structure matches with no name matches |
| let read_schema = union_schema(vec![ |
| Schema::Null, |
| point_3d_no_default_schema(), |
| point_2d_schema(), |
| point_3d_schema(), |
| ]); |
| assert!(!SchemaCompatibility::can_read( |
| &point_2d_fullname_schema(), |
| &read_schema |
| )); |
| } |
| |
| #[test] |
| fn test_union_resolution_first_structure_match_3d() { |
| // multiple structure matches with no name matches |
| let read_schema = union_schema(vec![ |
| Schema::Null, |
| point_3d_no_default_schema(), |
| point_3d_schema(), |
| point_2d_schema(), |
| ]); |
| assert!(!SchemaCompatibility::can_read( |
| &point_2d_fullname_schema(), |
| &read_schema |
| )); |
| } |
| |
| #[test] |
| fn test_union_resolution_named_structure_match() { |
| // multiple structure matches with a short name match |
| let read_schema = union_schema(vec![ |
| Schema::Null, |
| point_2d_schema(), |
| point_3d_match_name_schema(), |
| point_3d_schema(), |
| ]); |
| assert!(!SchemaCompatibility::can_read( |
| &point_2d_fullname_schema(), |
| &read_schema |
| )); |
| } |
| |
| #[test] |
| fn test_union_resolution_full_name_match() { |
| // there is a full name match that should be chosen |
| let read_schema = union_schema(vec![ |
| Schema::Null, |
| point_2d_schema(), |
| point_3d_match_name_schema(), |
| point_3d_schema(), |
| point_2d_fullname_schema(), |
| ]); |
| assert!(SchemaCompatibility::can_read( |
| &point_2d_fullname_schema(), |
| &read_schema |
| )); |
| } |
| } |
