From 038050f8a0e613868f677959d1e82b92cd55668c Mon Sep 17 00:00:00 2001 From: allenyuchen Date: Wed, 25 Mar 2026 17:45:50 -0700 Subject: [PATCH 1/6] feat: Support flatten Option to UnionSchema --- avro/src/schema/union.rs | 13 +- avro/src/serde/deser_schema/enums.rs | 35 +- avro/src/serde/deser_schema/mod.rs | 111 ++++- avro/src/serde/ser_schema/mod.rs | 37 +- avro/src/serde/ser_schema/union.rs | 24 +- avro/tests/nullable_union.rs | 595 +++++++++++++++++++++++++++ avro/tests/union_schema.rs | 12 +- 7 files changed, 774 insertions(+), 53 deletions(-) create mode 100644 avro/tests/nullable_union.rs diff --git a/avro/src/schema/union.rs b/avro/src/schema/union.rs index ca9a997b..4216a8a1 100644 --- a/avro/src/schema/union.rs +++ b/avro/src/schema/union.rs @@ -109,8 +109,7 @@ impl UnionSchema { name: &str, names: &'s HashMap>, ) -> Result, Error> { - for index in self.named_index.iter().copied() { - let schema = &self.schemas[index]; + for (index, schema) in self.schemas.iter().enumerate() { if let Some(schema_name) = schema.name() && schema_name.name() == name { @@ -123,6 +122,16 @@ impl UnionSchema { schema }; return Ok(Some((index, schema))); + } else if schema.to_string() == name { + let schema = if let Schema::Ref { name } = schema { + names + .get(name) + .ok_or_else(|| Details::SchemaResolutionError(name.clone()))? + .borrow() + } else { + schema + }; + return Ok(Some((index, schema))); } } Ok(None) diff --git a/avro/src/serde/deser_schema/enums.rs b/avro/src/serde/deser_schema/enums.rs index bba48fdd..dfd80c37 100644 --- a/avro/src/serde/deser_schema/enums.rs +++ b/avro/src/serde/deser_schema/enums.rs @@ -15,18 +15,17 @@ // specific language governing permissions and limitations // under the License. -use std::{borrow::Borrow, io::Read}; - use serde::{ Deserializer, de::{DeserializeSeed, EnumAccess, Unexpected, VariantAccess, Visitor}, }; +use std::{borrow::{Borrow, Cow}, io::Read}; use super::{Config, DESERIALIZE_ANY, SchemaAwareDeserializer, identifier::IdentifierDeserializer}; use crate::{ Error, Schema, error::Details, - schema::{EnumSchema, UnionSchema}, + schema::EnumSchema, util::zag_i32, }; @@ -100,17 +99,21 @@ impl<'de, 's, 'r, R: Read> VariantAccess<'de> for PlainEnumDeserializer<'s, 'r, pub struct UnionEnumDeserializer<'s, 'r, R: Read, S: Borrow> { reader: &'r mut R, - variants: &'s [Schema], + schema: &'s Schema, config: Config<'s, S>, } impl<'s, 'r, R: Read, S: Borrow> UnionEnumDeserializer<'s, 'r, R, S> { - pub fn new(reader: &'r mut R, schema: &'s UnionSchema, config: Config<'s, S>) -> Self { - Self { + pub fn new( + reader: &'r mut R, + schema: &'s Schema, + config: Config<'s, S>, + ) -> Result { + Ok(Self { reader, - variants: schema.variants(), + schema, config, - } + }) } } @@ -124,20 +127,16 @@ impl<'de, 's, 'r, R: Read, S: Borrow> EnumAccess<'de> where V: DeserializeSeed<'de>, { - let index = zag_i32(self.reader)?; - let index = usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))?; - let schema = self.variants.get(index).ok_or(Details::GetUnionVariant { - index: index as i64, - num_variants: self.variants.len(), - })?; - + let name = match self.schema.name() { + Some(name) => Cow::Borrowed(name.name()), + None => Cow::Owned(self.schema.to_string()), + }; Ok(( - seed.deserialize(IdentifierDeserializer::index(index as u32))?, - UnionVariantAccess::new(schema, self.reader, self.config)?, + seed.deserialize(IdentifierDeserializer::string(&name))?, + UnionVariantAccess::new(self.schema, self.reader, self.config)?, )) } } - pub struct UnionVariantAccess<'s, 'r, R: Read, S: Borrow> { schema: &'s Schema, reader: &'r mut R, diff --git a/avro/src/serde/deser_schema/mod.rs b/avro/src/serde/deser_schema/mod.rs index 45287a39..f1363894 100644 --- a/avro/src/serde/deser_schema/mod.rs +++ b/avro/src/serde/deser_schema/mod.rs @@ -35,11 +35,10 @@ mod tuple; use block::BlockDeserializer; use enums::PlainEnumDeserializer; +use enums::UnionEnumDeserializer; use record::RecordDeserializer; use tuple::{ManyTupleDeserializer, OneTupleDeserializer}; -use crate::serde::deser_schema::enums::UnionEnumDeserializer; - /// Configure the deserializer. #[derive(Debug)] pub struct Config<'s, S: Borrow> { @@ -129,6 +128,12 @@ impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { Ok(self) } + fn with_nullable_union_three_plus_variants( + self, + ) -> ThreePlusVariantUnionDeserializer<'s, 'r, R, S> { + ThreePlusVariantUnionDeserializer::new(self) + } + /// Read the union and create a new deserializer with the existing reader and config. /// /// This will resolve the read schema if it is a reference. @@ -539,7 +544,6 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> V: Visitor<'de>, { if let Schema::Union(union) = self.schema - && union.variants().len() == 2 && union.is_nullable() { let index = zag_i32(self.reader)?; @@ -548,7 +552,14 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> if let Schema::Null = schema { visitor.visit_none() } else { - visitor.visit_some(self.with_different_schema(schema)?) + if union.variants().len() == 2 { + visitor.visit_some(self.with_different_schema(schema)?) + } else { + visitor.visit_some( + self.with_different_schema(schema)? + .with_nullable_union_three_plus_variants(), + ) + } } } else { Err(self.error("option", "Expected Schema::Union([Schema::Null, _])")) @@ -724,7 +735,15 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> visitor.visit_enum(PlainEnumDeserializer::new(self.reader, schema)) } Schema::Union(union) => { - visitor.visit_enum(UnionEnumDeserializer::new(self.reader, union, self.config)) + let index = zag_i32(self.reader)?; + let index = + usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))?; + let schema = union.get_variant(index)?; + visitor.visit_enum(UnionEnumDeserializer::new( + self.reader, + schema, + self.config, + )?) } _ => Err(self.error("enum", "Expected Schema::Enum | Schema::Union")), } @@ -753,6 +772,88 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> } } +struct ThreePlusVariantUnionDeserializer<'s, 'r, R: Read, S: Borrow> { + inner: SchemaAwareDeserializer<'s, 'r, R, S>, +} + +impl<'s, 'r, R: Read, S: Borrow> ThreePlusVariantUnionDeserializer<'s, 'r, R, S> { + fn new(inner: SchemaAwareDeserializer<'s, 'r, R, S>) -> Self { + Self { inner } + } +} + +macro_rules! forward_to_inner_deserializer { + ($( $method:ident($($arg:ident: $arg_ty:ty),*); )*) => { + $( + fn $method(self, $($arg: $arg_ty,)* visitor: V) -> Result + where + V: Visitor<'de>, + { + self.inner.$method($($arg,)* visitor) + } + )* + }; +} + +impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> + for ThreePlusVariantUnionDeserializer<'s, 'r, R, S> +{ + type Error = Error; + + fn deserialize_enum( + self, + _name: &'static str, + _variants: &'static [&'static str], + visitor: V, + ) -> Result + where + V: Visitor<'de>, + { + visitor.visit_enum(UnionEnumDeserializer::new( + self.inner.reader, + self.inner.schema, + self.inner.config, + )?) + } + + fn is_human_readable(&self) -> bool { + self.inner.config.human_readable + } + + forward_to_inner_deserializer! { + deserialize_any(); + deserialize_bool(); + deserialize_i8(); + deserialize_i16(); + deserialize_i32(); + deserialize_i64(); + deserialize_i128(); + deserialize_u8(); + deserialize_u16(); + deserialize_u32(); + deserialize_u64(); + deserialize_u128(); + deserialize_f32(); + deserialize_f64(); + deserialize_char(); + deserialize_str(); + deserialize_string(); + deserialize_bytes(); + deserialize_byte_buf(); + deserialize_option(); + deserialize_unit(); + deserialize_seq(); + deserialize_map(); + deserialize_identifier(); + deserialize_ignored_any(); + deserialize_unit_struct(name: &'static str); + deserialize_newtype_struct(name: &'static str); + deserialize_tuple(len: usize); + deserialize_tuple_struct(name: &'static str, len: usize); + deserialize_struct(name: &'static str, fields: &'static [&'static str]); + } +} + #[cfg(test)] mod tests { use std::fmt::Debug; diff --git a/avro/src/serde/ser_schema/mod.rs b/avro/src/serde/ser_schema/mod.rs index 5aba68b1..39b16981 100644 --- a/avro/src/serde/ser_schema/mod.rs +++ b/avro/src/serde/ser_schema/mod.rs @@ -374,7 +374,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' fn serialize_none(self) -> Result { if let Schema::Union(union) = self.schema - && union.variants().len() == 2 + // && union.variants().len() == 2 && let Some(null_index) = union.index_of_schema_kind(SchemaKind::Null) { zig_i32(null_index as i32, &mut *self.writer) @@ -387,15 +387,19 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' where T: ?Sized + Serialize, { - if let Schema::Union(union) = self.schema - && union.variants().len() == 2 - && let Some(null_index) = union.index_of_schema_kind(SchemaKind::Null) - { - let some_index = (null_index + 1) & 1; - let mut bytes_written = zig_i32(some_index as i32, &mut *self.writer)?; - bytes_written += - value.serialize(self.with_different_schema(&union.variants()[some_index])?)?; - Ok(bytes_written) + if let Schema::Union(union) = self.schema { + if union.variants().len() == 2 + && let Some(null_index) = union.index_of_schema_kind(SchemaKind::Null) + { + let some_index = (null_index + 1) & 1; + let mut bytes_written = zig_i32(some_index as i32, &mut *self.writer)?; + bytes_written += + value.serialize(self.with_different_schema(&union.variants()[some_index])?)?; + Ok(bytes_written) + } else { + let union_serializer = UnionSerializer::new(self.writer, union, self.config); + value.serialize(union_serializer) + } } else { Err(self.error("some", "Expected Schema::Union([Schema::Null, _])")) } @@ -1919,29 +1923,28 @@ mod tests { } #[derive(Serialize)] - #[serde(untagged)] enum InnerUnion { - IntField(i32), + Int(i32), } let rs = ResolvedSchema::try_from(&schema)?; // Flattening a Option into the underlying union is NOT supported let null_record = TestRecord { inner_union: None }; - assert_serialize_err( + assert_serialize( null_record, &schema, rs.get_names(), - r#"Failed to serialize field 'innerUnion' of record RecordSchema { name: Name { name: "TestRecord", .. }, fields: [RecordField { name: "innerUnion", schema: Union(UnionSchema { schemas: [Null, Record(RecordSchema { name: Name { name: "innerRecordFoo", .. }, fields: [RecordField { name: "foo", schema: String, .. }], .. }), Record(RecordSchema { name: Name { name: "innerRecordBar", .. }, fields: [RecordField { name: "bar", schema: String, .. }], .. }), Int, String] }), .. }], .. }: Failed to serialize value of type `none`: Expected Schema::Union([Schema::Null, _])"#, + &[0], ); let foo_record = TestRecord { - inner_union: Some(InnerUnion::IntField(42)), + inner_union: Some(InnerUnion::Int(42)), }; - assert_serialize_err( + assert_serialize( foo_record, &schema, rs.get_names(), - r#"Failed to serialize field 'innerUnion' of record RecordSchema { name: Name { name: "TestRecord", .. }, fields: [RecordField { name: "innerUnion", schema: Union(UnionSchema { schemas: [Null, Record(RecordSchema { name: Name { name: "innerRecordFoo", .. }, fields: [RecordField { name: "foo", schema: String, .. }], .. }), Record(RecordSchema { name: Name { name: "innerRecordBar", .. }, fields: [RecordField { name: "bar", schema: String, .. }], .. }), Int, String] }), .. }], .. }: Failed to serialize value of type `some`: Expected Schema::Union([Schema::Null, _])"#, + &[6, 84], ); Ok(()) } diff --git a/avro/src/serde/ser_schema/union.rs b/avro/src/serde/ser_schema/union.rs index 4c020111..7ec1e4df 100644 --- a/avro/src/serde/ser_schema/union.rs +++ b/avro/src/serde/ser_schema/union.rs @@ -390,15 +390,29 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for UnionSerializer<'s, 'w, fn serialize_newtype_variant( self, - _: &'static str, - _: u32, - _: &'static str, - _: &T, + _name: &'static str, + _variant_index: u32, + variant: &'static str, + value: &T, ) -> Result where T: ?Sized + Serialize, { - Err(self.error("newtype variant", "Nested unions are not supported")) + match self.union.find_named_schema(variant, self.config.names)? { + Some((index, schema)) => { + let mut bytes_written = zig_i32(index as i32, &mut *self.writer)?; + bytes_written += value.serialize(SchemaAwareSerializer::new( + self.writer, + schema, + self.config, + )?)?; + Ok(bytes_written) + } + _ => Err(self.error( + "newtype variant", + format!("Expected Schema with name: {variant} in variants"), + )), + } } fn serialize_seq(self, len: Option) -> Result { diff --git a/avro/tests/nullable_union.rs b/avro/tests/nullable_union.rs new file mode 100644 index 00000000..b0c9818c --- /dev/null +++ b/avro/tests/nullable_union.rs @@ -0,0 +1,595 @@ +// 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. + +/* + Compiling apache-avro v0.22.0 (/home/coder/avro-rs/avro) + Finished `test` profile [unoptimized + debuginfo] target(s) in 0.62s +──────────── + Nextest run ID 4c2f8dd6-2ff4-4de0-b4b5-9a7709c6749b with nextest profile: default + Starting 36 tests across 1 binary + FAIL [ 0.005s] apache-avro::nullable_union nullable_enum::avro_json_encoding_compatible_my_enum_a + stdout ─── + + running 1 test + test nullable_enum::avro_json_encoding_compatible_my_enum_a ... FAILED + + failures: + + failures: + nullable_enum::avro_json_encoding_compatible_my_enum_a + + test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 35 filtered out; finished in 0.00s + + stderr ─── + + Backtrace omitted. Run with RUST_BACKTRACE=1 to display it. + Run with RUST_BACKTRACE=full to include source snippets. + + The application panicked (crashed). + apache_avro::error::Error: Error { details: Failed to serialize value of type `newtype variant` using Schema::Enum(EnumSchema { name: Name { name: "MyEnum", .. }, symbols: ["A", "B"], .. }): Expected Schema::Union } + in avro/tests/nullable_union.rs, line 214 + thread: nullable_enum::avro_json_encoding_compatible_my_enum_a + + PASS [ 0.004s] apache-avro::nullable_union nullable_enum::rusty_my_enum_a + PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_my_enum_a + PASS [ 0.004s] apache-avro::nullable_union nullable_enum::rusty_null + PASS [ 0.005s] apache-avro::nullable_union nullable_enum::avro_json_encoding_compatible_null + PASS [ 0.005s] apache-avro::nullable_union nullable_enum::null_variant_enum_my_enum_a + PASS [ 0.005s] apache-avro::nullable_union nullable_enum::null_variant_enum_null + PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_int_42 + PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_my_record_a_27 + PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::rusty_int_42 + PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::rusty_null + PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::rusty_my_enum_a + PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_my_record_a_27 + PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_null + PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_int_42 + PASS [ 0.006s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_my_enum_a + FAIL [ 0.004s] apache-avro::nullable_union nullable_primitive_int::avro_json_encoding_compatible_int_42 + stdout ─── + + running 1 test + test nullable_primitive_int::avro_json_encoding_compatible_int_42 ... FAILED + + failures: + + failures: + nullable_primitive_int::avro_json_encoding_compatible_int_42 + + test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 35 filtered out; finished in 0.00s + + stderr ─── + + Backtrace omitted. Run with RUST_BACKTRACE=1 to display it. + Run with RUST_BACKTRACE=full to include source snippets. + + The application panicked (crashed). + apache_avro::error::Error: Error { details: Failed to serialize value of type `newtype variant` using Schema::Int: Expected Schema::Union } + in avro/tests/nullable_union.rs, line 269 + thread: nullable_primitive_int::avro_json_encoding_compatible_int_42 + + PASS [ 0.004s] apache-avro::nullable_union nullable_primitive_int::null_variant_enum_null + PASS [ 0.004s] apache-avro::nullable_union nullable_primitive_int::null_variant_enum_int_42 + PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_my_record_a_27 + PASS [ 0.005s] apache-avro::nullable_union nullable_primitive_int::avro_json_encoding_compatible_null + PASS [ 0.007s] apache-avro::nullable_union nullable_primitive_int::rusty_int_42 + PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_null + PASS [ 0.004s] apache-avro::nullable_union nullable_primitive_int::rusty_null + FAIL [ 0.005s] apache-avro::nullable_union nullable_record::avro_json_encoding_compatible_my_record_a_27 + stdout ─── + + running 1 test + test nullable_record::avro_json_encoding_compatible_my_record_a_27 ... FAILED + + failures: + + failures: + nullable_record::avro_json_encoding_compatible_my_record_a_27 + + test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 35 filtered out; finished in 0.00s + + stderr ─── + + Backtrace omitted. Run with RUST_BACKTRACE=1 to display it. + Run with RUST_BACKTRACE=full to include source snippets. + + The application panicked (crashed). + apache_avro::error::Error: Error { details: Failed to serialize value of type `newtype variant` using Schema::Record(RecordSchema { name: Name { name: "MyRecord", .. }, fields: [RecordField { name: "a", schema: Int, .. }], .. }): Expected Schema::Union } + in avro/tests/nullable_union.rs, line 336 + thread: nullable_record::avro_json_encoding_compatible_my_record_a_27 + + PASS [ 0.005s] apache-avro::nullable_union nullable_record::rusty_my_record_a_27 + PASS [ 0.005s] apache-avro::nullable_union nullable_record::avro_json_encoding_compatible_null + PASS [ 0.007s] apache-avro::nullable_union nullable_record::null_variant_enum_my_record_a_27 + PASS [ 0.007s] apache-avro::nullable_union nullable_record::null_variant_enum_null + PASS [ 0.005s] apache-avro::nullable_union nullable_record::rusty_null + PASS [ 0.005s] apache-avro::nullable_union nullable_untagged_pitfall::null_variant_enum_my_record_b_27 + PASS [ 0.006s] apache-avro::nullable_union nullable_untagged_pitfall::null_variant_enum_my_record_a_27 + PASS [ 0.004s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_untagged_my_record_a_27 + PASS [ 0.005s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_my_record_a_27 + PASS [ 0.004s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_untagged_my_record_b_27 + PASS [ 0.005s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_my_record_b_27 +──────────── + Summary [ 0.027s] 36 tests run: 33 passed, 3 failed, 0 skipped + FAIL [ 0.005s] apache-avro::nullable_union nullable_enum::avro_json_encoding_compatible_my_enum_a + FAIL [ 0.004s] apache-avro::nullable_union nullable_primitive_int::avro_json_encoding_compatible_int_42 + FAIL [ 0.005s] apache-avro::nullable_union nullable_record::avro_json_encoding_compatible_my_record_a_27 +error: test run failed +*/ + +use std::fmt::Debug; + +use apache_avro::Schema; +use apache_avro::reader::datum::GenericDatumReader; +use apache_avro::writer::datum::GenericDatumWriter; +use apache_avro_test_helper::TestResult; +use pretty_assertions::assert_eq; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +#[track_caller] +fn assert_roundtrip(value: T, schema: &Schema) -> TestResult +where + T: Serialize + DeserializeOwned + PartialEq + Debug, +{ + let serialized = GenericDatumWriter::builder(schema) + .build()? + .write_ser_to_vec(&value)?; + let deserialized: T = GenericDatumReader::builder(schema) + .build()? + .read_deser(&mut &serialized[..])?; + + assert_eq!(deserialized, value); + Ok(()) +} + +mod nullable_enum { + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyEnum { + A, + B, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + MyEnum(MyEnum), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + MyEnum(MyEnum), + } + + fn schema() -> Schema { + Schema::parse_str( + r#" + [ + "null", + { + "type": "enum", + "name": "MyEnum", + "symbols": ["A", "B"] + } + ] + "#, + ) + .unwrap() + } + + #[test] + fn null_variant_enum_null() -> TestResult { + assert_roundtrip(MyUnionNullable::Null, &schema()) + } + + #[test] + fn rusty_null() -> TestResult { + assert_roundtrip(None::, &schema()) + } + + #[test] + fn avro_json_encoding_compatible_null() -> TestResult { + assert_roundtrip(None::, &schema()) + } + + #[test] + fn null_variant_enum_my_enum_a() -> TestResult { + assert_roundtrip(MyUnionNullable::MyEnum(MyEnum::A), &schema()) + } + + #[test] + fn rusty_my_enum_a() -> TestResult { + assert_roundtrip(Some(MyEnum::A), &schema()) + } + + #[test] + #[should_panic] + fn avro_json_encoding_compatible_my_enum_a() { + assert_roundtrip(Some(MyUnionAvroJsonEncoding::MyEnum(MyEnum::A)), &schema()).unwrap(); + } +} + +mod nullable_primitive_int { + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + Int(i32), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + #[serde(rename = "int")] + Int(i32), + } + + fn schema() -> Schema { + Schema::parse_str( + r#" + [ + "null", + "int" + ] + "#, + ) + .unwrap() + } + + #[test] + fn null_variant_enum_null() -> TestResult { + assert_roundtrip(MyUnionNullable::Null, &schema()) + } + + #[test] + fn rusty_null() -> TestResult { + assert_roundtrip(None::, &schema()) + } + + #[test] + fn avro_json_encoding_compatible_null() -> TestResult { + assert_roundtrip(None::, &schema()) + } + + #[test] + fn null_variant_enum_int_42() -> TestResult { + assert_roundtrip(MyUnionNullable::Int(42), &schema()) + } + + #[test] + fn rusty_int_42() -> TestResult { + assert_roundtrip(Some(42_i32), &schema()) + } + + #[test] + #[should_panic] + fn avro_json_encoding_compatible_int_42() { + assert_roundtrip(Some(MyUnionAvroJsonEncoding::Int(42)), &schema()).unwrap(); + } +} + +mod nullable_record { + use super::*; + + const NULLABLE_RECORD_SCHEMA: &str = r#" + [ + "null", + { + "type": "record", + "name": "MyRecord", + "fields": [ + {"name": "a", "type": "int"} + ] + } + ] + "#; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecord { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + MyRecord(MyRecord), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + MyRecord(MyRecord), + } + + fn schema() -> Schema { + Schema::parse_str(NULLABLE_RECORD_SCHEMA).unwrap() + } + + #[test] + fn null_variant_enum_null() -> TestResult { + assert_roundtrip(MyUnionNullable::Null, &schema()) + } + + #[test] + fn rusty_null() -> TestResult { + assert_roundtrip(None::, &schema()) + } + + #[test] + fn avro_json_encoding_compatible_null() -> TestResult { + assert_roundtrip(None::, &schema()) + } + + #[test] + fn null_variant_enum_my_record_a_27() -> TestResult { + assert_roundtrip(MyUnionNullable::MyRecord(MyRecord { a: 27 }), &schema()) + } + + #[test] + fn rusty_my_record_a_27() -> TestResult { + assert_roundtrip(Some(MyRecord { a: 27 }), &schema()) + } + + #[test] + #[should_panic] + fn avro_json_encoding_compatible_my_record_a_27() { + assert_roundtrip( + Some(MyUnionAvroJsonEncoding::MyRecord(MyRecord { a: 27 })), + &schema(), + ) + .unwrap(); + } +} + +mod nullable_int_enum_record { + use super::*; + + const NULLABLE_INT_ENUM_RECORD_SCHEMA: &str = r#" + [ + "null", + "int", + { + "type": "enum", + "name": "MyEnum", + "symbols": ["A", "B"] + }, + { + "type": "record", + "name": "MyRecord", + "fields": [ + {"name": "a", "type": "int"} + ] + } + ] + "#; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyEnum { + A, + B, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecord { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + Int(i32), + MyEnum(MyEnum), + MyRecord(MyRecord), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + #[serde(untagged)] + enum MyUnionUntagged { + Int(i32), + MyEnum(MyEnum), + MyRecord(MyRecord), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + Int(i32), + MyEnum(MyEnum), + MyRecord(MyRecord), + } + + fn schema() -> Schema { + Schema::parse_str(NULLABLE_INT_ENUM_RECORD_SCHEMA).unwrap() + } + + #[test] + fn null_variant_enum_null() -> TestResult { + assert_roundtrip(MyUnionNullable::Null, &schema()) + } + + #[test] + fn rusty_null() { + assert_roundtrip(None::, &schema()).unwrap(); + } + + #[test] + fn rusty_untagged_null() { + assert_roundtrip(None::, &schema()).unwrap(); + } + + #[test] + fn null_variant_enum_int_42() -> TestResult { + assert_roundtrip(MyUnionNullable::Int(42), &schema()) + } + + #[test] + fn rusty_int_42() { + assert_roundtrip(Some(MyUnionAvroJsonEncoding::Int(42)), &schema()).unwrap(); + } + + #[test] + fn rusty_untagged_int_42() { + assert_roundtrip(Some(MyUnionUntagged::Int(42)), &schema()).unwrap(); + } + + #[test] + fn null_variant_enum_my_enum_a() -> TestResult { + assert_roundtrip(MyUnionNullable::MyEnum(MyEnum::A), &schema()) + } + + #[test] + fn rusty_my_enum_a() { + assert_roundtrip(Some(MyUnionAvroJsonEncoding::MyEnum(MyEnum::A)), &schema()).unwrap(); + } + + #[test] + // Idk why this is erroring, the error source is from serde. However, I'm fine with not + // supporting this anyways since supporting untagged enums itself is opening a whole new can of + // worms + #[ignore] + fn rusty_untagged_my_enum_a() { + assert_roundtrip(Some(MyUnionUntagged::MyEnum(MyEnum::A)), &schema()).unwrap() + } + + #[test] + fn null_variant_enum_my_record_a_27() -> TestResult { + assert_roundtrip(MyUnionNullable::MyRecord(MyRecord { a: 27 }), &schema()) + } + + #[test] + fn rusty_my_record_a_27() { + assert_roundtrip( + Some(MyUnionAvroJsonEncoding::MyRecord(MyRecord { a: 27 })), + &schema(), + ) + .unwrap() + } + + #[test] + fn rusty_untagged_my_record_a_27() { + assert_roundtrip( + Some(MyUnionUntagged::MyRecord(MyRecord { a: 27 })), + &schema(), + ) + .unwrap(); + } +} + +mod nullable_untagged_pitfall { + use super::*; + + const NULLABLE_RECORD_SCHEMA: &str = r#" + [ + "null", + { + "type": "record", + "name": "MyRecordA", + "fields": [ + {"name": "a", "type": "int"} + ] + }, + { + "type": "record", + "name": "MyRecordB", + "fields": [ + {"name": "a", "type": "int"} + ] + } + ] + "#; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecordA { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecordB { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + MyRecordA(MyRecordA), + MyRecordB(MyRecordB), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + #[serde(untagged)] + enum MyUnionUntagged { + MyRecordA(MyRecordA), + MyRecordB(MyRecordB), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + MyRecordA(MyRecordA), + MyRecordB(MyRecordB), + } + + fn schema() -> Schema { + Schema::parse_str(NULLABLE_RECORD_SCHEMA).unwrap() + } + + #[test] + fn null_variant_enum_my_record_a_27() -> TestResult { + assert_roundtrip(MyUnionNullable::MyRecordA(MyRecordA { a: 27 }), &schema()) + } + + #[test] + fn rusty_my_record_a_27() { + assert_roundtrip( + Some(MyUnionAvroJsonEncoding::MyRecordA(MyRecordA { a: 27 })), + &schema(), + ) + .unwrap(); + } + + #[test] + fn rusty_untagged_my_record_a_27() { + assert_roundtrip( + Some(MyUnionUntagged::MyRecordA(MyRecordA { a: 27 })), + &schema(), + ) + .unwrap(); + } + + #[test] + fn null_variant_enum_my_record_b_27() -> TestResult { + assert_roundtrip(MyUnionNullable::MyRecordB(MyRecordB { a: 27 }), &schema()) + } + + #[test] + fn rusty_my_record_b_27() { + assert_roundtrip( + Some(MyUnionAvroJsonEncoding::MyRecordB(MyRecordB { a: 27 })), + &schema(), + ) + .unwrap(); + } + + #[test] + #[should_panic] + fn rusty_untagged_my_record_b_27() { + assert_roundtrip( + Some(MyUnionUntagged::MyRecordB(MyRecordB { a: 27 })), + &schema(), + ) + .unwrap(); + } +} diff --git a/avro/tests/union_schema.rs b/avro/tests/union_schema.rs index e1b7fa03..013d6c05 100644 --- a/avro/tests/union_schema.rs +++ b/avro/tests/union_schema.rs @@ -117,7 +117,7 @@ static SCHEMA_D_STR: &str = r#"{ #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] enum UnionNoneAB { - None, + Null, A(A), B(B), } @@ -140,7 +140,7 @@ fn test_avro_3901_union_schema_round_trip_null_at_start() -> TestResult { assert_roundtrip(&input, &schemata[2], &schemata)?; let input = D { - field_union: UnionNoneAB::None, + field_union: UnionNoneAB::Null, field_d: "fooyy".to_string(), }; assert_roundtrip(&input, &schemata[2], &schemata)?; @@ -166,7 +166,7 @@ static SCHEMA_E_STR: &str = r#"{ #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] enum UnionANoneB { A(A), - None, + Null, B(B), } @@ -188,7 +188,7 @@ fn test_avro_3901_union_schema_round_trip_with_out_of_order_null() -> TestResult assert_roundtrip(&input, &schemata[2], &schemata)?; let input = E { - field_union: UnionANoneB::None, + field_union: UnionANoneB::Null, field_e: "barme2".to_string(), }; assert_roundtrip(&input, &schemata[2], &schemata)?; @@ -215,7 +215,7 @@ static SCHEMA_F_STR: &str = r#"{ enum UnionABNone { A(A), B(B), - None, + Null, } #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] @@ -242,7 +242,7 @@ fn test_avro_3901_union_schema_round_trip_with_end_null() -> TestResult { assert_roundtrip(&input, &schemata[2], &schemata)?; let input = F { - field_union: UnionABNone::None, + field_union: UnionABNone::Null, field_f: "aoee2".to_string(), }; assert_roundtrip(&input, &schemata[2], &schemata)?; From 7ede6839f0dcadf18e203911e4ad96c9a9aa993c Mon Sep 17 00:00:00 2001 From: allenyuchen Date: Thu, 26 Mar 2026 15:39:17 -0700 Subject: [PATCH 2/6] feat: Use index bumping for flattening Option to Union Schema for ser --- avro/src/schema/union.rs | 13 +----- avro/src/serde/ser_schema/mod.rs | 63 +++++++++++++++++++----------- avro/src/serde/ser_schema/union.rs | 24 +++--------- 3 files changed, 48 insertions(+), 52 deletions(-) diff --git a/avro/src/schema/union.rs b/avro/src/schema/union.rs index 4216a8a1..ca9a997b 100644 --- a/avro/src/schema/union.rs +++ b/avro/src/schema/union.rs @@ -109,7 +109,8 @@ impl UnionSchema { name: &str, names: &'s HashMap>, ) -> Result, Error> { - for (index, schema) in self.schemas.iter().enumerate() { + for index in self.named_index.iter().copied() { + let schema = &self.schemas[index]; if let Some(schema_name) = schema.name() && schema_name.name() == name { @@ -122,16 +123,6 @@ impl UnionSchema { schema }; return Ok(Some((index, schema))); - } else if schema.to_string() == name { - let schema = if let Schema::Ref { name } = schema { - names - .get(name) - .ok_or_else(|| Details::SchemaResolutionError(name.clone()))? - .borrow() - } else { - schema - }; - return Ok(Some((index, schema))); } } Ok(None) diff --git a/avro/src/serde/ser_schema/mod.rs b/avro/src/serde/ser_schema/mod.rs index 39b16981..4eaeb02f 100644 --- a/avro/src/serde/ser_schema/mod.rs +++ b/avro/src/serde/ser_schema/mod.rs @@ -87,6 +87,7 @@ pub struct SchemaAwareSerializer<'s, 'w, W: Write, S: Borrow> { /// This schema is guaranteed to not be a [`Schema::Ref`]. schema: &'s Schema, config: Config<'s, S>, + null_variant_index: Option, } impl<'s, 'w, W: Write, S: Borrow> SchemaAwareSerializer<'s, 'w, W, S> { @@ -104,6 +105,7 @@ impl<'s, 'w, W: Write, S: Borrow> SchemaAwareSerializer<'s, 'w, W, S> { writer, schema, config, + null_variant_index: None, }) } @@ -127,13 +129,25 @@ impl<'s, 'w, W: Write, S: Borrow> SchemaAwareSerializer<'s, 'w, W, S> { Ok(self) } + fn with_null_variant_index(mut self, null_variant_index: usize) -> Self { + self.null_variant_index = Some(null_variant_index); + self + } + + fn get_resolved_branch_index(&self, variant_index: usize) -> usize { + match self.null_variant_index { + Some(null_index) if variant_index >= null_index => variant_index + 1, + _ => variant_index, + } + } + /// Get the schema at the given index of the union, resolving references. fn get_resolved_union_variant( &self, union: &'s UnionSchema, - index: u32, + index: usize, ) -> Result<&'s Schema, Error> { - match union.get_variant(index as usize)? { + match union.get_variant(index)? { Schema::Ref { name } => self.config.get_schema(name), schema => Ok(schema), } @@ -374,7 +388,6 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' fn serialize_none(self) -> Result { if let Schema::Union(union) = self.schema - // && union.variants().len() == 2 && let Some(null_index) = union.index_of_schema_kind(SchemaKind::Null) { zig_i32(null_index as i32, &mut *self.writer) @@ -387,18 +400,17 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' where T: ?Sized + Serialize, { - if let Schema::Union(union) = self.schema { - if union.variants().len() == 2 - && let Some(null_index) = union.index_of_schema_kind(SchemaKind::Null) - { + if let Schema::Union(union) = self.schema + && let Some(null_index) = union.index_of_schema_kind(SchemaKind::Null) + { + if union.variants().len() == 2 { let some_index = (null_index + 1) & 1; let mut bytes_written = zig_i32(some_index as i32, &mut *self.writer)?; bytes_written += value.serialize(self.with_different_schema(&union.variants()[some_index])?)?; Ok(bytes_written) } else { - let union_serializer = UnionSerializer::new(self.writer, union, self.config); - value.serialize(union_serializer) + value.serialize(self.with_null_variant_index(null_index)) } } else { Err(self.error("some", "Expected Schema::Union([Schema::Null, _])")) @@ -449,14 +461,17 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' Err(self.error("unit variant", format!(r#"Expected symbol "{variant}" at index {variant_index} in enum"#))) } } - Schema::Union(union) => match self.get_resolved_union_variant(union, variant_index)? { - // Bare union - Schema::Null => zig_i32(variant_index as i32, &mut *self.writer), - Schema::Record(record) if record.fields.is_empty() && record.name.name() == variant => { - // Union of records - zig_i32(variant_index as i32, &mut *self.writer) + Schema::Union(union) => { + let variant_index = self.get_resolved_branch_index(variant_index as usize); + match self.get_resolved_union_variant(union, variant_index)? { + // Bare union + Schema::Null => zig_i32(variant_index as i32, &mut *self.writer), + Schema::Record(record) if record.fields.is_empty() && record.name.name() == variant => { + // Union of records + zig_i32(variant_index as i32, &mut *self.writer) + } + _ => Err(self.error("unit variant", format!("Expected Schema::Null | Schema::Record(name: {variant}, fields: []) at index {variant_index} in the union"))), } - _ => Err(self.error("unit variant", format!("Expected Schema::Null | Schema::Record(name: {variant}, fields: []) at index {variant_index} in the union"))), } _ => Err(self.error("unit variant", format!("Expected Schema::Enum(symbols[{variant_index}] == {variant}) | Schema::Union(variants[{variant_index}] == Schema::Null | Schema::Record(name: {variant}, fields: []))"))), } @@ -494,6 +509,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' where T: ?Sized + Serialize, { + let variant_index = self.get_resolved_branch_index(variant_index as usize); match self.schema { Schema::Union(union) => match self.get_resolved_union_variant(union, variant_index)? { Schema::Record(record) @@ -587,6 +603,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' variant: &'static str, len: usize, ) -> Result { + let variant_index = self.get_resolved_branch_index(variant_index as usize); if let Schema::Union(union) = self.schema && let Schema::Record(record) = self.get_resolved_union_variant(union, variant_index)? && record.fields.len() == len @@ -661,6 +678,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' variant: &'static str, len: usize, ) -> Result { + let variant_index = self.get_resolved_branch_index(variant_index as usize); if let Schema::Union(union) = self.schema && let Schema::Record(record) = self.get_resolved_union_variant(union, variant_index)? && record.fields.len() == len @@ -1923,28 +1941,29 @@ mod tests { } #[derive(Serialize)] + #[serde(untagged)] enum InnerUnion { - Int(i32), + IntField(i32), } let rs = ResolvedSchema::try_from(&schema)?; // Flattening a Option into the underlying union is NOT supported let null_record = TestRecord { inner_union: None }; - assert_serialize( + assert_serialize_err( null_record, &schema, rs.get_names(), - &[0], + r#"Failed to serialize field 'innerUnion' of record RecordSchema { name: Name { name: "TestRecord", .. }, fields: [RecordField { name: "innerUnion", schema: Union(UnionSchema { schemas: [Null, Record(RecordSchema { name: Name { name: "innerRecordFoo", .. }, fields: [RecordField { name: "foo", schema: String, .. }], .. }), Record(RecordSchema { name: Name { name: "innerRecordBar", .. }, fields: [RecordField { name: "bar", schema: String, .. }], .. }), Int, String] }), .. }], .. }: Failed to serialize value of type `none`: Expected Schema::Union([Schema::Null, _])"#, ); let foo_record = TestRecord { - inner_union: Some(InnerUnion::Int(42)), + inner_union: Some(InnerUnion::IntField(42)), }; - assert_serialize( + assert_serialize_err( foo_record, &schema, rs.get_names(), - &[6, 84], + r#"Failed to serialize field 'innerUnion' of record RecordSchema { name: Name { name: "TestRecord", .. }, fields: [RecordField { name: "innerUnion", schema: Union(UnionSchema { schemas: [Null, Record(RecordSchema { name: Name { name: "innerRecordFoo", .. }, fields: [RecordField { name: "foo", schema: String, .. }], .. }), Record(RecordSchema { name: Name { name: "innerRecordBar", .. }, fields: [RecordField { name: "bar", schema: String, .. }], .. }), Int, String] }), .. }], .. }: Failed to serialize value of type `some`: Expected Schema::Union([Schema::Null, _])"#, ); Ok(()) } diff --git a/avro/src/serde/ser_schema/union.rs b/avro/src/serde/ser_schema/union.rs index 7ec1e4df..4c020111 100644 --- a/avro/src/serde/ser_schema/union.rs +++ b/avro/src/serde/ser_schema/union.rs @@ -390,29 +390,15 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for UnionSerializer<'s, 'w, fn serialize_newtype_variant( self, - _name: &'static str, - _variant_index: u32, - variant: &'static str, - value: &T, + _: &'static str, + _: u32, + _: &'static str, + _: &T, ) -> Result where T: ?Sized + Serialize, { - match self.union.find_named_schema(variant, self.config.names)? { - Some((index, schema)) => { - let mut bytes_written = zig_i32(index as i32, &mut *self.writer)?; - bytes_written += value.serialize(SchemaAwareSerializer::new( - self.writer, - schema, - self.config, - )?)?; - Ok(bytes_written) - } - _ => Err(self.error( - "newtype variant", - format!("Expected Schema with name: {variant} in variants"), - )), - } + Err(self.error("newtype variant", "Nested unions are not supported")) } fn serialize_seq(self, len: Option) -> Result { From 989b66e90561fd7043474e868f3647d0cfd277d4 Mon Sep 17 00:00:00 2001 From: allenyuchen Date: Thu, 26 Mar 2026 17:14:13 -0700 Subject: [PATCH 3/6] feat: Use index bumping for flatten Option for deser --- avro/src/serde/deser_schema/enums.rs | 45 +++++++--- avro/src/serde/deser_schema/mod.rs | 123 +++++---------------------- avro/src/serde/ser_schema/mod.rs | 48 +++++------ avro/tests/union_schema.rs | 12 +-- 4 files changed, 79 insertions(+), 149 deletions(-) diff --git a/avro/src/serde/deser_schema/enums.rs b/avro/src/serde/deser_schema/enums.rs index dfd80c37..61f18dcb 100644 --- a/avro/src/serde/deser_schema/enums.rs +++ b/avro/src/serde/deser_schema/enums.rs @@ -15,17 +15,18 @@ // specific language governing permissions and limitations // under the License. +use std::{borrow::Borrow, io::Read}; + use serde::{ Deserializer, de::{DeserializeSeed, EnumAccess, Unexpected, VariantAccess, Visitor}, }; -use std::{borrow::{Borrow, Cow}, io::Read}; use super::{Config, DESERIALIZE_ANY, SchemaAwareDeserializer, identifier::IdentifierDeserializer}; use crate::{ Error, Schema, error::Details, - schema::EnumSchema, + schema::{EnumSchema, UnionSchema}, util::zag_i32, }; @@ -99,21 +100,31 @@ impl<'de, 's, 'r, R: Read> VariantAccess<'de> for PlainEnumDeserializer<'s, 'r, pub struct UnionEnumDeserializer<'s, 'r, R: Read, S: Borrow> { reader: &'r mut R, - schema: &'s Schema, + variants: &'s [Schema], config: Config<'s, S>, + branch_index: Option, } impl<'s, 'r, R: Read, S: Borrow> UnionEnumDeserializer<'s, 'r, R, S> { pub fn new( reader: &'r mut R, - schema: &'s Schema, + schema: &'s UnionSchema, config: Config<'s, S>, - ) -> Result { - Ok(Self { + branch_index: Option, + ) -> Self { + Self { reader, - schema, + variants: schema.variants(), config, - }) + branch_index, + } + } + + fn get_variant_index(&self, branch_index: usize) -> usize { + match self.branch_index { + Some(null_index) if branch_index >= null_index => branch_index - 1, + _ => branch_index, + } } } @@ -127,13 +138,21 @@ impl<'de, 's, 'r, R: Read, S: Borrow> EnumAccess<'de> where V: DeserializeSeed<'de>, { - let name = match self.schema.name() { - Some(name) => Cow::Borrowed(name.name()), - None => Cow::Owned(self.schema.to_string()), + let index = match self.branch_index { + Some(index) => index, + None => { + let index = zag_i32(self.reader)?; + usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))? + } }; + let schema = self.variants.get(index).ok_or(Details::GetUnionVariant { + index: index as i64, + num_variants: self.variants.len(), + })?; + let variant_index = self.get_variant_index(index); Ok(( - seed.deserialize(IdentifierDeserializer::string(&name))?, - UnionVariantAccess::new(self.schema, self.reader, self.config)?, + seed.deserialize(IdentifierDeserializer::index(variant_index as u32))?, + UnionVariantAccess::new(schema, self.reader, self.config)?, )) } } diff --git a/avro/src/serde/deser_schema/mod.rs b/avro/src/serde/deser_schema/mod.rs index f1363894..f58cb1ed 100644 --- a/avro/src/serde/deser_schema/mod.rs +++ b/avro/src/serde/deser_schema/mod.rs @@ -78,6 +78,7 @@ pub struct SchemaAwareDeserializer<'s, 'r, R: Read, S: Borrow> { /// This schema is guaranteed to not be a [`Schema::Ref`]. schema: &'s Schema, config: Config<'s, S>, + branch_index: Option, } impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { @@ -95,12 +96,14 @@ impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { reader, schema, config, + branch_index: None, }) } else { Ok(Self { reader, schema, config, + branch_index: None, }) } } @@ -128,18 +131,22 @@ impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { Ok(self) } - fn with_nullable_union_three_plus_variants( - self, - ) -> ThreePlusVariantUnionDeserializer<'s, 'r, R, S> { - ThreePlusVariantUnionDeserializer::new(self) + fn with_branch_index(mut self, branch_index: usize) -> Self { + self.branch_index = Some(branch_index); + self } /// Read the union and create a new deserializer with the existing reader and config. /// /// This will resolve the read schema if it is a reference. fn with_union(self, schema: &'s UnionSchema) -> Result { - let index = zag_i32(self.reader)?; - let index = usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))?; + let index = match self.branch_index { + Some(index) => index, + None => { + let index = zag_i32(self.reader)?; + usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))? + } + }; let variant = schema.get_variant(index)?; self.with_different_schema(variant) } @@ -555,10 +562,7 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> if union.variants().len() == 2 { visitor.visit_some(self.with_different_schema(schema)?) } else { - visitor.visit_some( - self.with_different_schema(schema)? - .with_nullable_union_three_plus_variants(), - ) + visitor.visit_some(self.with_branch_index(index)) } } } else { @@ -734,17 +738,12 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> Schema::Enum(schema) => { visitor.visit_enum(PlainEnumDeserializer::new(self.reader, schema)) } - Schema::Union(union) => { - let index = zag_i32(self.reader)?; - let index = - usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))?; - let schema = union.get_variant(index)?; - visitor.visit_enum(UnionEnumDeserializer::new( - self.reader, - schema, - self.config, - )?) - } + Schema::Union(union) => visitor.visit_enum(UnionEnumDeserializer::new( + self.reader, + union, + self.config, + self.branch_index, + )), _ => Err(self.error("enum", "Expected Schema::Enum | Schema::Union")), } } @@ -772,88 +771,6 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> } } -struct ThreePlusVariantUnionDeserializer<'s, 'r, R: Read, S: Borrow> { - inner: SchemaAwareDeserializer<'s, 'r, R, S>, -} - -impl<'s, 'r, R: Read, S: Borrow> ThreePlusVariantUnionDeserializer<'s, 'r, R, S> { - fn new(inner: SchemaAwareDeserializer<'s, 'r, R, S>) -> Self { - Self { inner } - } -} - -macro_rules! forward_to_inner_deserializer { - ($( $method:ident($($arg:ident: $arg_ty:ty),*); )*) => { - $( - fn $method(self, $($arg: $arg_ty,)* visitor: V) -> Result - where - V: Visitor<'de>, - { - self.inner.$method($($arg,)* visitor) - } - )* - }; -} - -impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> - for ThreePlusVariantUnionDeserializer<'s, 'r, R, S> -{ - type Error = Error; - - fn deserialize_enum( - self, - _name: &'static str, - _variants: &'static [&'static str], - visitor: V, - ) -> Result - where - V: Visitor<'de>, - { - visitor.visit_enum(UnionEnumDeserializer::new( - self.inner.reader, - self.inner.schema, - self.inner.config, - )?) - } - - fn is_human_readable(&self) -> bool { - self.inner.config.human_readable - } - - forward_to_inner_deserializer! { - deserialize_any(); - deserialize_bool(); - deserialize_i8(); - deserialize_i16(); - deserialize_i32(); - deserialize_i64(); - deserialize_i128(); - deserialize_u8(); - deserialize_u16(); - deserialize_u32(); - deserialize_u64(); - deserialize_u128(); - deserialize_f32(); - deserialize_f64(); - deserialize_char(); - deserialize_str(); - deserialize_string(); - deserialize_bytes(); - deserialize_byte_buf(); - deserialize_option(); - deserialize_unit(); - deserialize_seq(); - deserialize_map(); - deserialize_identifier(); - deserialize_ignored_any(); - deserialize_unit_struct(name: &'static str); - deserialize_newtype_struct(name: &'static str); - deserialize_tuple(len: usize); - deserialize_tuple_struct(name: &'static str, len: usize); - deserialize_struct(name: &'static str, fields: &'static [&'static str]); - } -} - #[cfg(test)] mod tests { use std::fmt::Debug; diff --git a/avro/src/serde/ser_schema/mod.rs b/avro/src/serde/ser_schema/mod.rs index 4eaeb02f..9d4db3a4 100644 --- a/avro/src/serde/ser_schema/mod.rs +++ b/avro/src/serde/ser_schema/mod.rs @@ -134,7 +134,7 @@ impl<'s, 'w, W: Write, S: Borrow> SchemaAwareSerializer<'s, 'w, W, S> { self } - fn get_resolved_branch_index(&self, variant_index: usize) -> usize { + fn get_branch_index(&self, variant_index: usize) -> usize { match self.null_variant_index { Some(null_index) if variant_index >= null_index => variant_index + 1, _ => variant_index, @@ -462,15 +462,15 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' } } Schema::Union(union) => { - let variant_index = self.get_resolved_branch_index(variant_index as usize); - match self.get_resolved_union_variant(union, variant_index)? { + let branch_index = self.get_branch_index(variant_index as usize); + match self.get_resolved_union_variant(union, branch_index)? { // Bare union - Schema::Null => zig_i32(variant_index as i32, &mut *self.writer), + Schema::Null => zig_i32(branch_index as i32, &mut *self.writer), Schema::Record(record) if record.fields.is_empty() && record.name.name() == variant => { // Union of records - zig_i32(variant_index as i32, &mut *self.writer) + zig_i32(branch_index as i32, &mut *self.writer) } - _ => Err(self.error("unit variant", format!("Expected Schema::Null | Schema::Record(name: {variant}, fields: []) at index {variant_index} in the union"))), + _ => Err(self.error("unit variant", format!("Expected Schema::Null | Schema::Record(name: {variant}, fields: []) at index {branch_index} in the union"))), } } _ => Err(self.error("unit variant", format!("Expected Schema::Enum(symbols[{variant_index}] == {variant}) | Schema::Union(variants[{variant_index}] == Schema::Null | Schema::Record(name: {variant}, fields: []))"))), @@ -509,9 +509,9 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' where T: ?Sized + Serialize, { - let variant_index = self.get_resolved_branch_index(variant_index as usize); + let branch_index = self.get_branch_index(variant_index as usize); match self.schema { - Schema::Union(union) => match self.get_resolved_union_variant(union, variant_index)? { + Schema::Union(union) => match self.get_resolved_union_variant(union, branch_index)? { Schema::Record(record) if record.fields.len() == 1 && record.name.name() == variant @@ -521,13 +521,13 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' == Some(&Bool(true)) => { // Union of records - let mut bytes_written = zig_i32(variant_index as i32, &mut *self.writer)?; + let mut bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; let schema = &record.fields[0].schema; bytes_written += value.serialize(self.with_different_schema(schema)?)?; Ok(bytes_written) } schema => { - let mut bytes_written = zig_i32(variant_index as i32, &mut *self.writer)?; + let mut bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; bytes_written += value.serialize(self.with_different_schema(schema)?)?; Ok(bytes_written) } @@ -603,13 +603,13 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' variant: &'static str, len: usize, ) -> Result { - let variant_index = self.get_resolved_branch_index(variant_index as usize); + let branch_index = self.get_branch_index(variant_index as usize); if let Schema::Union(union) = self.schema - && let Schema::Record(record) = self.get_resolved_union_variant(union, variant_index)? + && let Schema::Record(record) = self.get_resolved_union_variant(union, branch_index)? && record.fields.len() == len && record.name.name() == variant { - let bytes_written = zig_i32(variant_index as i32, &mut *self.writer)?; + let bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; Ok(ManyTupleSerializer::new( self.writer, record, @@ -617,7 +617,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' Some(bytes_written), )) } else { - Err(self.error("tuple variant", format!("Expected Schema::Union(variants[{variant_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) + Err(self.error("tuple variant", format!("Expected Schema::Union(variants[{branch_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) } } @@ -678,13 +678,13 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' variant: &'static str, len: usize, ) -> Result { - let variant_index = self.get_resolved_branch_index(variant_index as usize); + let branch_index = self.get_branch_index(variant_index as usize); if let Schema::Union(union) = self.schema - && let Schema::Record(record) = self.get_resolved_union_variant(union, variant_index)? + && let Schema::Record(record) = self.get_resolved_union_variant(union, branch_index)? && record.fields.len() == len && record.name.name() == variant { - let bytes_written = zig_i32(variant_index as i32, &mut *self.writer)?; + let bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; Ok(RecordSerializer::new( self.writer, record, @@ -692,7 +692,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' Some(bytes_written), )) } else { - Err(self.error("struct variant", format!("Expected Schema::Union(variants[{variant_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) + Err(self.error("struct variant", format!("Expected Schema::Union(variants[{branch_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) } } @@ -1941,21 +1941,15 @@ mod tests { } #[derive(Serialize)] - #[serde(untagged)] enum InnerUnion { IntField(i32), } let rs = ResolvedSchema::try_from(&schema)?; - // Flattening a Option into the underlying union is NOT supported let null_record = TestRecord { inner_union: None }; - assert_serialize_err( - null_record, - &schema, - rs.get_names(), - r#"Failed to serialize field 'innerUnion' of record RecordSchema { name: Name { name: "TestRecord", .. }, fields: [RecordField { name: "innerUnion", schema: Union(UnionSchema { schemas: [Null, Record(RecordSchema { name: Name { name: "innerRecordFoo", .. }, fields: [RecordField { name: "foo", schema: String, .. }], .. }), Record(RecordSchema { name: Name { name: "innerRecordBar", .. }, fields: [RecordField { name: "bar", schema: String, .. }], .. }), Int, String] }), .. }], .. }: Failed to serialize value of type `none`: Expected Schema::Union([Schema::Null, _])"#, - ); + assert_serialize(null_record, &schema, rs.get_names(), &[0]); + // Incorrect order, needs to be the 3rd variant, not the first one. let foo_record = TestRecord { inner_union: Some(InnerUnion::IntField(42)), }; @@ -1963,7 +1957,7 @@ mod tests { foo_record, &schema, rs.get_names(), - r#"Failed to serialize field 'innerUnion' of record RecordSchema { name: Name { name: "TestRecord", .. }, fields: [RecordField { name: "innerUnion", schema: Union(UnionSchema { schemas: [Null, Record(RecordSchema { name: Name { name: "innerRecordFoo", .. }, fields: [RecordField { name: "foo", schema: String, .. }], .. }), Record(RecordSchema { name: Name { name: "innerRecordBar", .. }, fields: [RecordField { name: "bar", schema: String, .. }], .. }), Int, String] }), .. }], .. }: Failed to serialize value of type `some`: Expected Schema::Union([Schema::Null, _])"#, + r#"Failed to serialize field 'innerUnion' of record RecordSchema { name: Name { name: "TestRecord", .. }, fields: [RecordField { name: "innerUnion", schema: Union(UnionSchema { schemas: [Null, Record(RecordSchema { name: Name { name: "innerRecordFoo", .. }, fields: [RecordField { name: "foo", schema: String, .. }], .. }), Record(RecordSchema { name: Name { name: "innerRecordBar", .. }, fields: [RecordField { name: "bar", schema: String, .. }], .. }), Int, String] }), .. }], .. }: Failed to serialize value of type `i32`: Expected Schema::Int | Schema::Date | Schema::TimeMillis"#, ); Ok(()) } diff --git a/avro/tests/union_schema.rs b/avro/tests/union_schema.rs index 013d6c05..e1b7fa03 100644 --- a/avro/tests/union_schema.rs +++ b/avro/tests/union_schema.rs @@ -117,7 +117,7 @@ static SCHEMA_D_STR: &str = r#"{ #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] enum UnionNoneAB { - Null, + None, A(A), B(B), } @@ -140,7 +140,7 @@ fn test_avro_3901_union_schema_round_trip_null_at_start() -> TestResult { assert_roundtrip(&input, &schemata[2], &schemata)?; let input = D { - field_union: UnionNoneAB::Null, + field_union: UnionNoneAB::None, field_d: "fooyy".to_string(), }; assert_roundtrip(&input, &schemata[2], &schemata)?; @@ -166,7 +166,7 @@ static SCHEMA_E_STR: &str = r#"{ #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] enum UnionANoneB { A(A), - Null, + None, B(B), } @@ -188,7 +188,7 @@ fn test_avro_3901_union_schema_round_trip_with_out_of_order_null() -> TestResult assert_roundtrip(&input, &schemata[2], &schemata)?; let input = E { - field_union: UnionANoneB::Null, + field_union: UnionANoneB::None, field_e: "barme2".to_string(), }; assert_roundtrip(&input, &schemata[2], &schemata)?; @@ -215,7 +215,7 @@ static SCHEMA_F_STR: &str = r#"{ enum UnionABNone { A(A), B(B), - Null, + None, } #[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] @@ -242,7 +242,7 @@ fn test_avro_3901_union_schema_round_trip_with_end_null() -> TestResult { assert_roundtrip(&input, &schemata[2], &schemata)?; let input = F { - field_union: UnionABNone::Null, + field_union: UnionABNone::None, field_f: "aoee2".to_string(), }; assert_roundtrip(&input, &schemata[2], &schemata)?; From ffc430149a3ebdc2ebbb3bb92693ad2044bd4437 Mon Sep 17 00:00:00 2001 From: allenyuchen Date: Tue, 31 Mar 2026 11:55:36 -0700 Subject: [PATCH 4/6] feat: Support Option flattening for AvroSchemaComponent --- avro/src/serde/derive.rs | 66 ++++++++++++++++++++++++++++++++-------- 1 file changed, 53 insertions(+), 13 deletions(-) diff --git a/avro/src/serde/derive.rs b/avro/src/serde/derive.rs index 6d66fcec..070876ef 100644 --- a/avro/src/serde/derive.rs +++ b/avro/src/serde/derive.rs @@ -22,7 +22,10 @@ use std::{ use crate::{ Schema, - schema::{FixedSchema, Name, NamespaceRef, RecordField, RecordSchema, UnionSchema, UuidSchema}, + schema::{ + FixedSchema, Name, NamespaceRef, RecordField, RecordSchema, SchemaKind, UnionSchema, + UuidSchema, + }, }; /// Trait for types that serve as an Avro data model. @@ -587,11 +590,16 @@ where named_schemas: &mut HashSet, enclosing_namespace: NamespaceRef, ) -> Schema { - let variants = vec![ - Schema::Null, - T::get_schema_in_ctxt(named_schemas, enclosing_namespace), - ]; - + let variants = match T::get_schema_in_ctxt(named_schemas, enclosing_namespace) { + Schema::Union(union) if union.index_of_schema_kind(SchemaKind::Null).is_some() => { + union.schemas + } + Schema::Union(union) => vec![Schema::Null] + .into_iter() + .chain(union.schemas) + .collect(), + schema => vec![Schema::Null, schema], + }; Schema::Union( UnionSchema::new(variants).expect("Option must produce a valid (non-nested) union"), ) @@ -950,11 +958,12 @@ mod tests { use apache_avro_test_helper::TestResult; use crate::{ - AvroSchema, Schema, + AvroSchema, AvroSchemaComponent, Schema, reader::datum::GenericDatumReader, - schema::{FixedSchema, Name}, + schema::{FixedSchema, Name, NamespaceRef, UnionSchema}, writer::datum::GenericDatumWriter, }; + use std::collections::HashSet; #[test] fn avro_rs_401_str() -> TestResult { @@ -1070,11 +1079,13 @@ mod tests { } #[test] - #[should_panic( - expected = "Option must produce a valid (non-nested) union: Error { details: Unions may not directly contain a union }" - )] - fn avro_rs_489_option_option() { - >>::get_schema(); + fn avro_rs_489_option_option() -> TestResult { + let schema = >>::get_schema(); + assert_eq!( + schema, + Schema::Union(UnionSchema::new(vec![Schema::Null, Schema::Int])?) + ); + Ok(()) } #[test] @@ -1207,4 +1218,33 @@ mod tests { ); Ok(()) } + + #[test] + fn test_nullable_complex_union() -> TestResult { + let schema = Schema::parse_str(r#"["null", "int", "string"]"#)?; + + #[allow(dead_code)] + enum MyUnion { + Int(i32), + String(String), + } + + impl AvroSchemaComponent for MyUnion { + fn get_schema_in_ctxt( + named_schemas: &mut HashSet, + enclosing_namespace: NamespaceRef, + ) -> Schema { + let int_schema = i32::get_schema_in_ctxt(named_schemas, enclosing_namespace); + let string_schema = String::get_schema_in_ctxt(named_schemas, enclosing_namespace); + Schema::Union( + UnionSchema::new(vec![Schema::Null, int_schema, string_schema]) + .expect("Union must be valid"), + ) + } + } + + assert_eq!(schema, Option::::get_schema()); + + Ok(()) + } } From 7c80e15916bacc4598ffa0ef0612763a48d2ac78 Mon Sep 17 00:00:00 2001 From: allenyuchen Date: Thu, 2 Apr 2026 23:35:37 -0700 Subject: [PATCH 5/6] fix: Panic on nested Null unions for Option --- avro/src/serde/derive.rs | 21 +++++---------------- 1 file changed, 5 insertions(+), 16 deletions(-) diff --git a/avro/src/serde/derive.rs b/avro/src/serde/derive.rs index 070876ef..1064b92f 100644 --- a/avro/src/serde/derive.rs +++ b/avro/src/serde/derive.rs @@ -22,10 +22,7 @@ use std::{ use crate::{ Schema, - schema::{ - FixedSchema, Name, NamespaceRef, RecordField, RecordSchema, SchemaKind, UnionSchema, - UuidSchema, - }, + schema::{FixedSchema, Name, NamespaceRef, RecordField, RecordSchema, UnionSchema, UuidSchema}, }; /// Trait for types that serve as an Avro data model. @@ -591,9 +588,6 @@ where enclosing_namespace: NamespaceRef, ) -> Schema { let variants = match T::get_schema_in_ctxt(named_schemas, enclosing_namespace) { - Schema::Union(union) if union.index_of_schema_kind(SchemaKind::Null).is_some() => { - union.schemas - } Schema::Union(union) => vec![Schema::Null] .into_iter() .chain(union.schemas) @@ -1079,13 +1073,9 @@ mod tests { } #[test] - fn avro_rs_489_option_option() -> TestResult { - let schema = >>::get_schema(); - assert_eq!( - schema, - Schema::Union(UnionSchema::new(vec![Schema::Null, Schema::Int])?) - ); - Ok(()) + #[should_panic(expected = "Unions cannot contain duplicate types, found at least two Null")] + fn avro_rs_489_option_option() { + >>::get_schema(); } #[test] @@ -1237,8 +1227,7 @@ mod tests { let int_schema = i32::get_schema_in_ctxt(named_schemas, enclosing_namespace); let string_schema = String::get_schema_in_ctxt(named_schemas, enclosing_namespace); Schema::Union( - UnionSchema::new(vec![Schema::Null, int_schema, string_schema]) - .expect("Union must be valid"), + UnionSchema::new(vec![int_schema, string_schema]).expect("Union must be valid"), ) } } From ac61167c5d59bd49b9344f21dcfb39c5e9b4fda9 Mon Sep 17 00:00:00 2001 From: Kriskras99 Date: Tue, 14 Jul 2026 23:30:11 +0200 Subject: [PATCH 6/6] fix: Improve documentation and simplify code --- .../documentation/serde_data_model_to_avro.rs | 2 + avro/src/serde/derive.rs | 31 +- avro/src/serde/deser_schema/enums.rs | 31 +- avro/src/serde/deser_schema/mod.rs | 26 +- avro/src/serde/ser_schema/mod.rs | 94 +-- avro/tests/avro_rs_528_nullable_union.rs | 437 +++++++++++++ avro/tests/nullable_union.rs | 595 ------------------ 7 files changed, 538 insertions(+), 678 deletions(-) create mode 100644 avro/tests/avro_rs_528_nullable_union.rs delete mode 100644 avro/tests/nullable_union.rs diff --git a/avro/src/documentation/serde_data_model_to_avro.rs b/avro/src/documentation/serde_data_model_to_avro.rs index 7016dbf6..fd8fafbc 100644 --- a/avro/src/documentation/serde_data_model_to_avro.rs +++ b/avro/src/documentation/serde_data_model_to_avro.rs @@ -40,6 +40,8 @@ //! - **string** => [`Schema::String`] //! - **byte array** => [`Schema::Bytes`] or [`Schema::Fixed`] //! - **option** => [`Schema::Union([Schema::Null, _])`](crate::schema::Schema::Union) +//! - If the schema of `T` is also a union schema and does not have a null variant, the schemas +//! are allowed to be merged. This results in a "flattened" `Option`. //! - **unit** => [`Schema::Null`] //! - **unit struct** => [`Schema::Record`] with the unqualified name equal to the name of the struct and zero fields //! - **unit variant** => See [Enums](#enums) diff --git a/avro/src/serde/derive.rs b/avro/src/serde/derive.rs index 1064b92f..9dbf5711 100644 --- a/avro/src/serde/derive.rs +++ b/avro/src/serde/derive.rs @@ -583,15 +583,25 @@ impl AvroSchemaComponent for Option where T: AvroSchemaComponent, { + /// The schema is a [`Schema::Union`] with the first variant set to [`Schema::Null`]. + /// + /// If the schema of `T` is a union, the variants of this union will be appended. If the `T` union + /// already contains a `Schema::Null` the construction will panic. + /// If the schema of `T` is not a union, it will be appended to the union. + /// + /// # Panics + /// If `T::get_schema_in_ctxt` returns a [`Schema::Union`] where one variant is [`Schema::Null`]. fn get_schema_in_ctxt( named_schemas: &mut HashSet, enclosing_namespace: NamespaceRef, ) -> Schema { let variants = match T::get_schema_in_ctxt(named_schemas, enclosing_namespace) { - Schema::Union(union) => vec![Schema::Null] - .into_iter() - .chain(union.schemas) - .collect(), + Schema::Union(mut union) => { + // It would be more efficient to append the null schema, but the (de)serializers have + // a fast path if the first variant is the null schema + union.schemas.insert(0, Schema::Null); + union.schemas + } schema => vec![Schema::Null, schema], }; Schema::Union( @@ -954,7 +964,7 @@ mod tests { use crate::{ AvroSchema, AvroSchemaComponent, Schema, reader::datum::GenericDatumReader, - schema::{FixedSchema, Name, NamespaceRef, UnionSchema}, + schema::{FixedSchema, Name, NamespaceRef}, writer::datum::GenericDatumWriter, }; use std::collections::HashSet; @@ -1220,15 +1230,8 @@ mod tests { } impl AvroSchemaComponent for MyUnion { - fn get_schema_in_ctxt( - named_schemas: &mut HashSet, - enclosing_namespace: NamespaceRef, - ) -> Schema { - let int_schema = i32::get_schema_in_ctxt(named_schemas, enclosing_namespace); - let string_schema = String::get_schema_in_ctxt(named_schemas, enclosing_namespace); - Schema::Union( - UnionSchema::new(vec![int_schema, string_schema]).expect("Union must be valid"), - ) + fn get_schema_in_ctxt(_: &mut HashSet, _: NamespaceRef) -> Schema { + Schema::union(vec![Schema::Int, Schema::String]).expect("Union must be valid") } } diff --git a/avro/src/serde/deser_schema/enums.rs b/avro/src/serde/deser_schema/enums.rs index 61f18dcb..c8973f06 100644 --- a/avro/src/serde/deser_schema/enums.rs +++ b/avro/src/serde/deser_schema/enums.rs @@ -102,7 +102,8 @@ pub struct UnionEnumDeserializer<'s, 'r, R: Read, S: Borrow> { reader: &'r mut R, variants: &'s [Schema], config: Config<'s, S>, - branch_index: Option, + /// The index of the null that belongs to the Option schema and has already been read. + flattened_option_null_index: Option, } impl<'s, 'r, R: Read, S: Borrow> UnionEnumDeserializer<'s, 'r, R, S> { @@ -110,20 +111,22 @@ impl<'s, 'r, R: Read, S: Borrow> UnionEnumDeserializer<'s, 'r, R, S> { reader: &'r mut R, schema: &'s UnionSchema, config: Config<'s, S>, - branch_index: Option, + flattened_option_null_index: Option, ) -> Self { Self { reader, variants: schema.variants(), config, - branch_index, + flattened_option_null_index, } } - fn get_variant_index(&self, branch_index: usize) -> usize { - match self.branch_index { - Some(null_index) if branch_index >= null_index => branch_index - 1, - _ => branch_index, + fn correct_index_for_serde(&self, serde_index: usize) -> usize { + match self.flattened_option_null_index { + // The index from Serde needs to be corrected for the flattened Option null schema, + // as Serde is not aware of the flattening. + Some(null_index) if serde_index >= null_index => serde_index - 1, + _ => serde_index, } } } @@ -138,24 +141,24 @@ impl<'de, 's, 'r, R: Read, S: Borrow> EnumAccess<'de> where V: DeserializeSeed<'de>, { - let index = match self.branch_index { - Some(index) => index, - None => { - let index = zag_i32(self.reader)?; - usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))? - } + let index = if let Some(index) = self.flattened_option_null_index { + index + } else { + let index = zag_i32(self.reader)?; + usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))? }; let schema = self.variants.get(index).ok_or(Details::GetUnionVariant { index: index as i64, num_variants: self.variants.len(), })?; - let variant_index = self.get_variant_index(index); + let variant_index = self.correct_index_for_serde(index); Ok(( seed.deserialize(IdentifierDeserializer::index(variant_index as u32))?, UnionVariantAccess::new(schema, self.reader, self.config)?, )) } } + pub struct UnionVariantAccess<'s, 'r, R: Read, S: Borrow> { schema: &'s Schema, reader: &'r mut R, diff --git a/avro/src/serde/deser_schema/mod.rs b/avro/src/serde/deser_schema/mod.rs index f58cb1ed..60fd991b 100644 --- a/avro/src/serde/deser_schema/mod.rs +++ b/avro/src/serde/deser_schema/mod.rs @@ -78,7 +78,7 @@ pub struct SchemaAwareDeserializer<'s, 'r, R: Read, S: Borrow> { /// This schema is guaranteed to not be a [`Schema::Ref`]. schema: &'s Schema, config: Config<'s, S>, - branch_index: Option, + flattened_option_null_index: Option, } impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { @@ -96,14 +96,14 @@ impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { reader, schema, config, - branch_index: None, + flattened_option_null_index: None, }) } else { Ok(Self { reader, schema, config, - branch_index: None, + flattened_option_null_index: None, }) } } @@ -131,8 +131,9 @@ impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { Ok(self) } - fn with_branch_index(mut self, branch_index: usize) -> Self { - self.branch_index = Some(branch_index); + /// Create a new deserializer which is aware that the Option has already read the union index. + fn with_flattened_option_null_index(mut self, index: usize) -> Self { + self.flattened_option_null_index = Some(index); self } @@ -140,12 +141,11 @@ impl<'s, 'r, R: Read, S: Borrow> SchemaAwareDeserializer<'s, 'r, R, S> { /// /// This will resolve the read schema if it is a reference. fn with_union(self, schema: &'s UnionSchema) -> Result { - let index = match self.branch_index { - Some(index) => index, - None => { - let index = zag_i32(self.reader)?; - usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))? - } + let index = if let Some(index) = self.flattened_option_null_index { + index + } else { + let index = zag_i32(self.reader)?; + usize::try_from(index).map_err(|e| Details::ConvertI32ToUsize(e, index))? }; let variant = schema.get_variant(index)?; self.with_different_schema(variant) @@ -562,7 +562,7 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> if union.variants().len() == 2 { visitor.visit_some(self.with_different_schema(schema)?) } else { - visitor.visit_some(self.with_branch_index(index)) + visitor.visit_some(self.with_flattened_option_null_index(index)) } } } else { @@ -742,7 +742,7 @@ impl<'de, 's, 'r, R: Read, S: Borrow> Deserializer<'de> self.reader, union, self.config, - self.branch_index, + self.flattened_option_null_index, )), _ => Err(self.error("enum", "Expected Schema::Enum | Schema::Union")), } diff --git a/avro/src/serde/ser_schema/mod.rs b/avro/src/serde/ser_schema/mod.rs index 9d4db3a4..7f3d1b3a 100644 --- a/avro/src/serde/ser_schema/mod.rs +++ b/avro/src/serde/ser_schema/mod.rs @@ -87,7 +87,7 @@ pub struct SchemaAwareSerializer<'s, 'w, W: Write, S: Borrow> { /// This schema is guaranteed to not be a [`Schema::Ref`]. schema: &'s Schema, config: Config<'s, S>, - null_variant_index: Option, + flattened_option_null_index: Option, } impl<'s, 'w, W: Write, S: Borrow> SchemaAwareSerializer<'s, 'w, W, S> { @@ -105,7 +105,7 @@ impl<'s, 'w, W: Write, S: Borrow> SchemaAwareSerializer<'s, 'w, W, S> { writer, schema, config, - null_variant_index: None, + flattened_option_null_index: None, }) } @@ -129,13 +129,16 @@ impl<'s, 'w, W: Write, S: Borrow> SchemaAwareSerializer<'s, 'w, W, S> { Ok(self) } - fn with_null_variant_index(mut self, null_variant_index: usize) -> Self { - self.null_variant_index = Some(null_variant_index); + /// Create a new serializer that is aware that the index of the null variant has been consumed by the Option. + fn with_flattened_option_null_index(mut self, index: usize) -> Self { + self.flattened_option_null_index = Some(index); self } - fn get_branch_index(&self, variant_index: usize) -> usize { - match self.null_variant_index { + fn correct_index_from_serde(&self, variant_index: usize) -> usize { + match self.flattened_option_null_index { + // The index from Serde needs to be corrected for the flattened Option null schema, + // as Serde is not aware of the flattening. Some(null_index) if variant_index >= null_index => variant_index + 1, _ => variant_index, } @@ -410,7 +413,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' value.serialize(self.with_different_schema(&union.variants()[some_index])?)?; Ok(bytes_written) } else { - value.serialize(self.with_null_variant_index(null_index)) + value.serialize(self.with_flattened_option_null_index(null_index)) } } else { Err(self.error("some", "Expected Schema::Union([Schema::Null, _])")) @@ -444,7 +447,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' fn serialize_unit_variant( self, - _name: &'static str, + name: &'static str, variant_index: u32, variant: &'static str, ) -> Result { @@ -462,15 +465,18 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' } } Schema::Union(union) => { - let branch_index = self.get_branch_index(variant_index as usize); - match self.get_resolved_union_variant(union, branch_index)? { + let corrected_index = self.correct_index_from_serde(variant_index as usize); + match self.get_resolved_union_variant(union, corrected_index)? { // Bare union - Schema::Null => zig_i32(branch_index as i32, &mut *self.writer), + Schema::Null => zig_i32(corrected_index as i32, &mut *self.writer), Schema::Record(record) if record.fields.is_empty() && record.name.name() == variant => { // Union of records - zig_i32(branch_index as i32, &mut *self.writer) + zig_i32(corrected_index as i32, &mut *self.writer) + } + _ if let Some((_, Schema::Enum(_))) = union.find_named_schema(name, self.config.names)? => { + Err(self.error("unit variant", format!("Expected Schema::Null | Schema::Record(name: {variant}, fields: []) at index {corrected_index} in the union. If the type has a plain enum inside a untagged enum, this is not supported by Serde."))) } - _ => Err(self.error("unit variant", format!("Expected Schema::Null | Schema::Record(name: {variant}, fields: []) at index {branch_index} in the union"))), + _ => Err(self.error("unit variant", format!("Expected Schema::Null | Schema::Record(name: {variant}, fields: []) at index {corrected_index} in the union"))), } } _ => Err(self.error("unit variant", format!("Expected Schema::Enum(symbols[{variant_index}] == {variant}) | Schema::Union(variants[{variant_index}] == Schema::Null | Schema::Record(name: {variant}, fields: []))"))), @@ -509,29 +515,31 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' where T: ?Sized + Serialize, { - let branch_index = self.get_branch_index(variant_index as usize); + let corrected_index = self.correct_index_from_serde(variant_index as usize); match self.schema { - Schema::Union(union) => match self.get_resolved_union_variant(union, branch_index)? { - Schema::Record(record) - if record.fields.len() == 1 - && record.name.name() == variant - && record - .attributes - .get("org.apache.avro.rust.union_of_records") - == Some(&Bool(true)) => - { - // Union of records - let mut bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; - let schema = &record.fields[0].schema; - bytes_written += value.serialize(self.with_different_schema(schema)?)?; - Ok(bytes_written) - } - schema => { - let mut bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; - bytes_written += value.serialize(self.with_different_schema(schema)?)?; - Ok(bytes_written) + Schema::Union(union) => { + match self.get_resolved_union_variant(union, corrected_index)? { + Schema::Record(record) + if record.fields.len() == 1 + && record.name.name() == variant + && record + .attributes + .get("org.apache.avro.rust.union_of_records") + == Some(&Bool(true)) => + { + // Union of records + let mut bytes_written = zig_i32(corrected_index as i32, &mut *self.writer)?; + let schema = &record.fields[0].schema; + bytes_written += value.serialize(self.with_different_schema(schema)?)?; + Ok(bytes_written) + } + schema => { + let mut bytes_written = zig_i32(corrected_index as i32, &mut *self.writer)?; + bytes_written += value.serialize(self.with_different_schema(schema)?)?; + Ok(bytes_written) + } } - }, + } _ => Err(self.error("newtype variant", "Expected Schema::Union")), } } @@ -603,13 +611,14 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' variant: &'static str, len: usize, ) -> Result { - let branch_index = self.get_branch_index(variant_index as usize); + let corrected_index = self.correct_index_from_serde(variant_index as usize); if let Schema::Union(union) = self.schema - && let Schema::Record(record) = self.get_resolved_union_variant(union, branch_index)? + && let Schema::Record(record) = + self.get_resolved_union_variant(union, corrected_index)? && record.fields.len() == len && record.name.name() == variant { - let bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; + let bytes_written = zig_i32(corrected_index as i32, &mut *self.writer)?; Ok(ManyTupleSerializer::new( self.writer, record, @@ -617,7 +626,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' Some(bytes_written), )) } else { - Err(self.error("tuple variant", format!("Expected Schema::Union(variants[{branch_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) + Err(self.error("tuple variant", format!("Expected Schema::Union(variants[{corrected_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) } } @@ -678,13 +687,14 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' variant: &'static str, len: usize, ) -> Result { - let branch_index = self.get_branch_index(variant_index as usize); + let corrected_index = self.correct_index_from_serde(variant_index as usize); if let Schema::Union(union) = self.schema - && let Schema::Record(record) = self.get_resolved_union_variant(union, branch_index)? + && let Schema::Record(record) = + self.get_resolved_union_variant(union, corrected_index)? && record.fields.len() == len && record.name.name() == variant { - let bytes_written = zig_i32(branch_index as i32, &mut *self.writer)?; + let bytes_written = zig_i32(corrected_index as i32, &mut *self.writer)?; Ok(RecordSerializer::new( self.writer, record, @@ -692,7 +702,7 @@ impl<'s, 'w, W: Write, S: Borrow> Serializer for SchemaAwareSerializer<' Some(bytes_written), )) } else { - Err(self.error("struct variant", format!("Expected Schema::Union(variants[{branch_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) + Err(self.error("struct variant", format!("Expected Schema::Union(variants[{corrected_index}] == Schema::Record(name: {variant}, fields.len() == {len}))"))) } } diff --git a/avro/tests/avro_rs_528_nullable_union.rs b/avro/tests/avro_rs_528_nullable_union.rs new file mode 100644 index 00000000..86327acd --- /dev/null +++ b/avro/tests/avro_rs_528_nullable_union.rs @@ -0,0 +1,437 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::fmt::Debug; + +use apache_avro::Schema; +use apache_avro::reader::datum::GenericDatumReader; +use apache_avro::writer::datum::GenericDatumWriter; +use pretty_assertions::assert_eq; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +#[track_caller] +fn assert_roundtrip(value: &T, schema: &Schema) +where + T: Serialize + DeserializeOwned + PartialEq + Debug, +{ + let serialized = GenericDatumWriter::builder(schema) + .build() + .unwrap() + .write_ser_to_vec(&value) + .unwrap(); + let deserialized: T = GenericDatumReader::builder(schema) + .build() + .unwrap() + .read_deser(&mut &serialized[..]) + .unwrap(); + + assert_eq!(&deserialized, value); +} + +mod nullable_enum { + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyEnum { + A, + B, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + MyEnum(MyEnum), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + MyEnum(MyEnum), + } + + fn schema() -> Schema { + Schema::parse_str( + r#" + [ + "null", + { + "type": "enum", + "name": "MyEnum", + "symbols": ["A", "B"] + } + ]"#, + ) + .unwrap() + } + + #[test] + fn null_variant_enum() { + let schema = schema(); + assert_roundtrip(&MyUnionNullable::Null, &schema); + assert_roundtrip(&MyUnionNullable::MyEnum(MyEnum::A), &schema); + assert_roundtrip(&MyUnionNullable::MyEnum(MyEnum::B), &schema); + } + + #[test] + fn option_enum() { + let schema = schema(); + assert_roundtrip(&None::, &schema); + assert_roundtrip(&Some(MyEnum::A), &schema); + assert_roundtrip(&Some(MyEnum::B), &schema); + } + + #[test] + fn avro_json_encoding_compatible_null() { + assert_roundtrip(&None::, &schema()); + } + + #[test] + // TODO: A (union) enum with only one variant is incorrectly seen as an option by the serializer + // and fails to serialize + fn avro_json_encoding_compatible_my_enum_a() { + // I think this should work + assert_roundtrip(&Some(MyUnionAvroJsonEncoding::MyEnum(MyEnum::A)), &schema()); + } +} + +mod nullable_primitive_int { + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + Int(i32), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + #[serde(rename = "int")] + Int(i32), + } + + fn schema() -> Schema { + Schema::parse_str( + r#" + [ + "null", + "int" + ] + "#, + ) + .unwrap() + } + + #[test] + fn null_variant_enum() { + let schema = schema(); + assert_roundtrip(&MyUnionNullable::Null, &schema); + assert_roundtrip(&MyUnionNullable::Int(42), &schema); + } + + #[test] + fn option_i32() { + let schema = schema(); + assert_roundtrip(&None::, &schema); + assert_roundtrip(&Some(42_i32), &schema); + } + + #[test] + fn avro_json_encoding_compatible_null() { + assert_roundtrip(&None::, &schema()); + } + + #[test] + // TODO: A (union) enum with only one variant is incorrectly seen as an option by the serializer + // and fails to serialize + fn avro_json_encoding_compatible_int_42() { + assert_roundtrip(&Some(MyUnionAvroJsonEncoding::Int(42)), &schema()); + } +} + +mod nullable_record { + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecord { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + MyRecord(MyRecord), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + MyRecord(MyRecord), + } + + fn schema() -> Schema { + Schema::parse_str( + r#" + [ + "null", + { + "type": "record", + "name": "MyRecord", + "fields": [ + {"name": "a", "type": "int"} + ] + } + ] + "#, + ) + .unwrap() + } + + #[test] + fn null_variant_enum() { + let schema = schema(); + assert_roundtrip(&MyUnionNullable::Null, &schema); + assert_roundtrip(&MyUnionNullable::MyRecord(MyRecord { a: 27 }), &schema); + } + + #[test] + fn option_record() { + let schema = schema(); + assert_roundtrip(&None::, &schema); + assert_roundtrip(&Some(MyRecord { a: 27 }), &schema); + } + + #[test] + fn avro_json_encoding_compatible_null() { + assert_roundtrip(&None::, &schema()); + } + + #[test] + // TODO: A (union) enum with only one variant is incorrectly seen as an option by the serializer + // and fails to serialize + fn avro_json_encoding_compatible_my_record_a_27() { + assert_roundtrip( + &Some(MyUnionAvroJsonEncoding::MyRecord(MyRecord { a: 27 })), + &schema(), + ); + } +} + +mod nullable_int_enum_record { + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyEnum { + A, + B, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecord { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + Int(i32), + MyEnum(MyEnum), + MyRecord(MyRecord), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + #[serde(untagged)] + enum MyUnionUntagged { + Int(i32), + MyEnum(MyEnum), + MyRecord(MyRecord), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + Int(i32), + MyEnum(MyEnum), + MyRecord(MyRecord), + } + + fn schema() -> Schema { + Schema::parse_str( + r#" + [ + "null", + "int", + { + "type": "enum", + "name": "MyEnum", + "symbols": ["A", "B"] + }, + { + "type": "record", + "name": "MyRecord", + "fields": [ + {"name": "a", "type": "int"} + ] + } + ] + "#, + ) + .unwrap() + } + + #[test] + fn null_variant_enum() { + let schema = schema(); + assert_roundtrip(&MyUnionNullable::Null, &schema); + assert_roundtrip(&MyUnionNullable::Int(42), &schema); + assert_roundtrip(&MyUnionNullable::MyEnum(MyEnum::A), &schema); + assert_roundtrip(&MyUnionNullable::MyEnum(MyEnum::B), &schema); + assert_roundtrip(&MyUnionNullable::MyRecord(MyRecord { a: 27 }), &schema); + } + + #[test] + fn option_enum() { + let schema = schema(); + assert_roundtrip(&None::, &schema); + assert_roundtrip(&Some(MyUnionAvroJsonEncoding::Int(42)), &schema); + assert_roundtrip(&Some(MyUnionAvroJsonEncoding::MyEnum(MyEnum::A)), &schema); + assert_roundtrip( + &Some(MyUnionAvroJsonEncoding::MyRecord(MyRecord { a: 27 })), + &schema, + ); + } + + #[test] + fn option_enum_untagged() { + let schema = schema(); + assert_roundtrip(&None::, &schema); + assert_roundtrip(&Some(MyUnionUntagged::Int(42)), &schema); + assert_roundtrip( + &Some(MyUnionUntagged::MyRecord(MyRecord { a: 27 })), + &schema, + ); + } + + #[test] + #[should_panic( + expected = "If the type has a plain enum inside a untagged enum, this is not supported by Serde." + )] + fn rusty_untagged_my_enum_a() { + // This cannot work. Because the parent enum is untagged, the serializer will get the index + // of the child enum. This makes it impossible for the serializer to pick the right variant. + assert_roundtrip(&Some(MyUnionUntagged::MyEnum(MyEnum::A)), &schema()); + } +} + +mod nullable_untagged_pitfall { + use super::*; + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecordA { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + struct MyRecordB { + a: i32, + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionNullable { + Null, + MyRecordA(MyRecordA), + MyRecordB(MyRecordB), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + #[serde(untagged)] + enum MyUnionUntagged { + MyRecordA(MyRecordA), + MyRecordB(MyRecordB), + } + + #[derive(Debug, Serialize, Deserialize, PartialEq)] + enum MyUnionAvroJsonEncoding { + MyRecordA(MyRecordA), + MyRecordB(MyRecordB), + } + + fn schema() -> Schema { + Schema::parse_str( + r#" + [ + "null", + { + "type": "record", + "name": "MyRecordA", + "fields": [ + {"name": "a", "type": "int"} + ] + }, + { + "type": "record", + "name": "MyRecordB", + "fields": [ + {"name": "a", "type": "int"} + ] + } + ] + "#, + ) + .unwrap() + } + + #[test] + fn null_variant_enum_my_record_a_27() { + let schema = schema(); + assert_roundtrip(&MyUnionNullable::Null, &schema); + assert_roundtrip(&MyUnionNullable::MyRecordA(MyRecordA { a: 27 }), &schema); + assert_roundtrip(&MyUnionNullable::MyRecordB(MyRecordB { a: 27 }), &schema); + } + + #[test] + fn rusty_my_record_a_27() { + let schema = schema(); + assert_roundtrip(&None::, &schema); + assert_roundtrip( + &Some(MyUnionAvroJsonEncoding::MyRecordA(MyRecordA { a: 27 })), + &schema, + ); + assert_roundtrip( + &Some(MyUnionAvroJsonEncoding::MyRecordB(MyRecordB { a: 27 })), + &schema, + ); + } + + #[test] + fn rusty_untagged_my_record_a_27() { + let schema = schema(); + assert_roundtrip(&None::, &schema); + assert_roundtrip( + &Some(MyUnionUntagged::MyRecordA(MyRecordA { a: 27 })), + &schema, + ); + } + + #[test] + #[should_panic(expected = "assertion failed: `(left == right)`")] + fn rusty_untagged_my_record_b_27() { + // Because the untagged enum has two exactly the same fields, this will be correctly serialized + // as MyRecordB, but incorrectly deserialized as MyRecordA. This is a limitation of Serde untagged. + assert_roundtrip( + &Some(MyUnionUntagged::MyRecordB(MyRecordB { a: 27 })), + &schema(), + ); + } +} diff --git a/avro/tests/nullable_union.rs b/avro/tests/nullable_union.rs deleted file mode 100644 index b0c9818c..00000000 --- a/avro/tests/nullable_union.rs +++ /dev/null @@ -1,595 +0,0 @@ -// 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. - -/* - Compiling apache-avro v0.22.0 (/home/coder/avro-rs/avro) - Finished `test` profile [unoptimized + debuginfo] target(s) in 0.62s -──────────── - Nextest run ID 4c2f8dd6-2ff4-4de0-b4b5-9a7709c6749b with nextest profile: default - Starting 36 tests across 1 binary - FAIL [ 0.005s] apache-avro::nullable_union nullable_enum::avro_json_encoding_compatible_my_enum_a - stdout ─── - - running 1 test - test nullable_enum::avro_json_encoding_compatible_my_enum_a ... FAILED - - failures: - - failures: - nullable_enum::avro_json_encoding_compatible_my_enum_a - - test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 35 filtered out; finished in 0.00s - - stderr ─── - - Backtrace omitted. Run with RUST_BACKTRACE=1 to display it. - Run with RUST_BACKTRACE=full to include source snippets. - - The application panicked (crashed). - apache_avro::error::Error: Error { details: Failed to serialize value of type `newtype variant` using Schema::Enum(EnumSchema { name: Name { name: "MyEnum", .. }, symbols: ["A", "B"], .. }): Expected Schema::Union } - in avro/tests/nullable_union.rs, line 214 - thread: nullable_enum::avro_json_encoding_compatible_my_enum_a - - PASS [ 0.004s] apache-avro::nullable_union nullable_enum::rusty_my_enum_a - PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_my_enum_a - PASS [ 0.004s] apache-avro::nullable_union nullable_enum::rusty_null - PASS [ 0.005s] apache-avro::nullable_union nullable_enum::avro_json_encoding_compatible_null - PASS [ 0.005s] apache-avro::nullable_union nullable_enum::null_variant_enum_my_enum_a - PASS [ 0.005s] apache-avro::nullable_union nullable_enum::null_variant_enum_null - PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_int_42 - PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_my_record_a_27 - PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::rusty_int_42 - PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::rusty_null - PASS [ 0.004s] apache-avro::nullable_union nullable_int_enum_record::rusty_my_enum_a - PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_my_record_a_27 - PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::null_variant_enum_null - PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_int_42 - PASS [ 0.006s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_my_enum_a - FAIL [ 0.004s] apache-avro::nullable_union nullable_primitive_int::avro_json_encoding_compatible_int_42 - stdout ─── - - running 1 test - test nullable_primitive_int::avro_json_encoding_compatible_int_42 ... FAILED - - failures: - - failures: - nullable_primitive_int::avro_json_encoding_compatible_int_42 - - test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 35 filtered out; finished in 0.00s - - stderr ─── - - Backtrace omitted. Run with RUST_BACKTRACE=1 to display it. - Run with RUST_BACKTRACE=full to include source snippets. - - The application panicked (crashed). - apache_avro::error::Error: Error { details: Failed to serialize value of type `newtype variant` using Schema::Int: Expected Schema::Union } - in avro/tests/nullable_union.rs, line 269 - thread: nullable_primitive_int::avro_json_encoding_compatible_int_42 - - PASS [ 0.004s] apache-avro::nullable_union nullable_primitive_int::null_variant_enum_null - PASS [ 0.004s] apache-avro::nullable_union nullable_primitive_int::null_variant_enum_int_42 - PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_my_record_a_27 - PASS [ 0.005s] apache-avro::nullable_union nullable_primitive_int::avro_json_encoding_compatible_null - PASS [ 0.007s] apache-avro::nullable_union nullable_primitive_int::rusty_int_42 - PASS [ 0.005s] apache-avro::nullable_union nullable_int_enum_record::rusty_untagged_null - PASS [ 0.004s] apache-avro::nullable_union nullable_primitive_int::rusty_null - FAIL [ 0.005s] apache-avro::nullable_union nullable_record::avro_json_encoding_compatible_my_record_a_27 - stdout ─── - - running 1 test - test nullable_record::avro_json_encoding_compatible_my_record_a_27 ... FAILED - - failures: - - failures: - nullable_record::avro_json_encoding_compatible_my_record_a_27 - - test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 35 filtered out; finished in 0.00s - - stderr ─── - - Backtrace omitted. Run with RUST_BACKTRACE=1 to display it. - Run with RUST_BACKTRACE=full to include source snippets. - - The application panicked (crashed). - apache_avro::error::Error: Error { details: Failed to serialize value of type `newtype variant` using Schema::Record(RecordSchema { name: Name { name: "MyRecord", .. }, fields: [RecordField { name: "a", schema: Int, .. }], .. }): Expected Schema::Union } - in avro/tests/nullable_union.rs, line 336 - thread: nullable_record::avro_json_encoding_compatible_my_record_a_27 - - PASS [ 0.005s] apache-avro::nullable_union nullable_record::rusty_my_record_a_27 - PASS [ 0.005s] apache-avro::nullable_union nullable_record::avro_json_encoding_compatible_null - PASS [ 0.007s] apache-avro::nullable_union nullable_record::null_variant_enum_my_record_a_27 - PASS [ 0.007s] apache-avro::nullable_union nullable_record::null_variant_enum_null - PASS [ 0.005s] apache-avro::nullable_union nullable_record::rusty_null - PASS [ 0.005s] apache-avro::nullable_union nullable_untagged_pitfall::null_variant_enum_my_record_b_27 - PASS [ 0.006s] apache-avro::nullable_union nullable_untagged_pitfall::null_variant_enum_my_record_a_27 - PASS [ 0.004s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_untagged_my_record_a_27 - PASS [ 0.005s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_my_record_a_27 - PASS [ 0.004s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_untagged_my_record_b_27 - PASS [ 0.005s] apache-avro::nullable_union nullable_untagged_pitfall::rusty_my_record_b_27 -──────────── - Summary [ 0.027s] 36 tests run: 33 passed, 3 failed, 0 skipped - FAIL [ 0.005s] apache-avro::nullable_union nullable_enum::avro_json_encoding_compatible_my_enum_a - FAIL [ 0.004s] apache-avro::nullable_union nullable_primitive_int::avro_json_encoding_compatible_int_42 - FAIL [ 0.005s] apache-avro::nullable_union nullable_record::avro_json_encoding_compatible_my_record_a_27 -error: test run failed -*/ - -use std::fmt::Debug; - -use apache_avro::Schema; -use apache_avro::reader::datum::GenericDatumReader; -use apache_avro::writer::datum::GenericDatumWriter; -use apache_avro_test_helper::TestResult; -use pretty_assertions::assert_eq; -use serde::de::DeserializeOwned; -use serde::{Deserialize, Serialize}; - -#[track_caller] -fn assert_roundtrip(value: T, schema: &Schema) -> TestResult -where - T: Serialize + DeserializeOwned + PartialEq + Debug, -{ - let serialized = GenericDatumWriter::builder(schema) - .build()? - .write_ser_to_vec(&value)?; - let deserialized: T = GenericDatumReader::builder(schema) - .build()? - .read_deser(&mut &serialized[..])?; - - assert_eq!(deserialized, value); - Ok(()) -} - -mod nullable_enum { - use super::*; - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyEnum { - A, - B, - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionNullable { - Null, - MyEnum(MyEnum), - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionAvroJsonEncoding { - MyEnum(MyEnum), - } - - fn schema() -> Schema { - Schema::parse_str( - r#" - [ - "null", - { - "type": "enum", - "name": "MyEnum", - "symbols": ["A", "B"] - } - ] - "#, - ) - .unwrap() - } - - #[test] - fn null_variant_enum_null() -> TestResult { - assert_roundtrip(MyUnionNullable::Null, &schema()) - } - - #[test] - fn rusty_null() -> TestResult { - assert_roundtrip(None::, &schema()) - } - - #[test] - fn avro_json_encoding_compatible_null() -> TestResult { - assert_roundtrip(None::, &schema()) - } - - #[test] - fn null_variant_enum_my_enum_a() -> TestResult { - assert_roundtrip(MyUnionNullable::MyEnum(MyEnum::A), &schema()) - } - - #[test] - fn rusty_my_enum_a() -> TestResult { - assert_roundtrip(Some(MyEnum::A), &schema()) - } - - #[test] - #[should_panic] - fn avro_json_encoding_compatible_my_enum_a() { - assert_roundtrip(Some(MyUnionAvroJsonEncoding::MyEnum(MyEnum::A)), &schema()).unwrap(); - } -} - -mod nullable_primitive_int { - use super::*; - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionNullable { - Null, - Int(i32), - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionAvroJsonEncoding { - #[serde(rename = "int")] - Int(i32), - } - - fn schema() -> Schema { - Schema::parse_str( - r#" - [ - "null", - "int" - ] - "#, - ) - .unwrap() - } - - #[test] - fn null_variant_enum_null() -> TestResult { - assert_roundtrip(MyUnionNullable::Null, &schema()) - } - - #[test] - fn rusty_null() -> TestResult { - assert_roundtrip(None::, &schema()) - } - - #[test] - fn avro_json_encoding_compatible_null() -> TestResult { - assert_roundtrip(None::, &schema()) - } - - #[test] - fn null_variant_enum_int_42() -> TestResult { - assert_roundtrip(MyUnionNullable::Int(42), &schema()) - } - - #[test] - fn rusty_int_42() -> TestResult { - assert_roundtrip(Some(42_i32), &schema()) - } - - #[test] - #[should_panic] - fn avro_json_encoding_compatible_int_42() { - assert_roundtrip(Some(MyUnionAvroJsonEncoding::Int(42)), &schema()).unwrap(); - } -} - -mod nullable_record { - use super::*; - - const NULLABLE_RECORD_SCHEMA: &str = r#" - [ - "null", - { - "type": "record", - "name": "MyRecord", - "fields": [ - {"name": "a", "type": "int"} - ] - } - ] - "#; - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - struct MyRecord { - a: i32, - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionNullable { - Null, - MyRecord(MyRecord), - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionAvroJsonEncoding { - MyRecord(MyRecord), - } - - fn schema() -> Schema { - Schema::parse_str(NULLABLE_RECORD_SCHEMA).unwrap() - } - - #[test] - fn null_variant_enum_null() -> TestResult { - assert_roundtrip(MyUnionNullable::Null, &schema()) - } - - #[test] - fn rusty_null() -> TestResult { - assert_roundtrip(None::, &schema()) - } - - #[test] - fn avro_json_encoding_compatible_null() -> TestResult { - assert_roundtrip(None::, &schema()) - } - - #[test] - fn null_variant_enum_my_record_a_27() -> TestResult { - assert_roundtrip(MyUnionNullable::MyRecord(MyRecord { a: 27 }), &schema()) - } - - #[test] - fn rusty_my_record_a_27() -> TestResult { - assert_roundtrip(Some(MyRecord { a: 27 }), &schema()) - } - - #[test] - #[should_panic] - fn avro_json_encoding_compatible_my_record_a_27() { - assert_roundtrip( - Some(MyUnionAvroJsonEncoding::MyRecord(MyRecord { a: 27 })), - &schema(), - ) - .unwrap(); - } -} - -mod nullable_int_enum_record { - use super::*; - - const NULLABLE_INT_ENUM_RECORD_SCHEMA: &str = r#" - [ - "null", - "int", - { - "type": "enum", - "name": "MyEnum", - "symbols": ["A", "B"] - }, - { - "type": "record", - "name": "MyRecord", - "fields": [ - {"name": "a", "type": "int"} - ] - } - ] - "#; - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyEnum { - A, - B, - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - struct MyRecord { - a: i32, - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionNullable { - Null, - Int(i32), - MyEnum(MyEnum), - MyRecord(MyRecord), - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - #[serde(untagged)] - enum MyUnionUntagged { - Int(i32), - MyEnum(MyEnum), - MyRecord(MyRecord), - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionAvroJsonEncoding { - Int(i32), - MyEnum(MyEnum), - MyRecord(MyRecord), - } - - fn schema() -> Schema { - Schema::parse_str(NULLABLE_INT_ENUM_RECORD_SCHEMA).unwrap() - } - - #[test] - fn null_variant_enum_null() -> TestResult { - assert_roundtrip(MyUnionNullable::Null, &schema()) - } - - #[test] - fn rusty_null() { - assert_roundtrip(None::, &schema()).unwrap(); - } - - #[test] - fn rusty_untagged_null() { - assert_roundtrip(None::, &schema()).unwrap(); - } - - #[test] - fn null_variant_enum_int_42() -> TestResult { - assert_roundtrip(MyUnionNullable::Int(42), &schema()) - } - - #[test] - fn rusty_int_42() { - assert_roundtrip(Some(MyUnionAvroJsonEncoding::Int(42)), &schema()).unwrap(); - } - - #[test] - fn rusty_untagged_int_42() { - assert_roundtrip(Some(MyUnionUntagged::Int(42)), &schema()).unwrap(); - } - - #[test] - fn null_variant_enum_my_enum_a() -> TestResult { - assert_roundtrip(MyUnionNullable::MyEnum(MyEnum::A), &schema()) - } - - #[test] - fn rusty_my_enum_a() { - assert_roundtrip(Some(MyUnionAvroJsonEncoding::MyEnum(MyEnum::A)), &schema()).unwrap(); - } - - #[test] - // Idk why this is erroring, the error source is from serde. However, I'm fine with not - // supporting this anyways since supporting untagged enums itself is opening a whole new can of - // worms - #[ignore] - fn rusty_untagged_my_enum_a() { - assert_roundtrip(Some(MyUnionUntagged::MyEnum(MyEnum::A)), &schema()).unwrap() - } - - #[test] - fn null_variant_enum_my_record_a_27() -> TestResult { - assert_roundtrip(MyUnionNullable::MyRecord(MyRecord { a: 27 }), &schema()) - } - - #[test] - fn rusty_my_record_a_27() { - assert_roundtrip( - Some(MyUnionAvroJsonEncoding::MyRecord(MyRecord { a: 27 })), - &schema(), - ) - .unwrap() - } - - #[test] - fn rusty_untagged_my_record_a_27() { - assert_roundtrip( - Some(MyUnionUntagged::MyRecord(MyRecord { a: 27 })), - &schema(), - ) - .unwrap(); - } -} - -mod nullable_untagged_pitfall { - use super::*; - - const NULLABLE_RECORD_SCHEMA: &str = r#" - [ - "null", - { - "type": "record", - "name": "MyRecordA", - "fields": [ - {"name": "a", "type": "int"} - ] - }, - { - "type": "record", - "name": "MyRecordB", - "fields": [ - {"name": "a", "type": "int"} - ] - } - ] - "#; - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - struct MyRecordA { - a: i32, - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - struct MyRecordB { - a: i32, - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionNullable { - Null, - MyRecordA(MyRecordA), - MyRecordB(MyRecordB), - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - #[serde(untagged)] - enum MyUnionUntagged { - MyRecordA(MyRecordA), - MyRecordB(MyRecordB), - } - - #[derive(Debug, Serialize, Deserialize, PartialEq)] - enum MyUnionAvroJsonEncoding { - MyRecordA(MyRecordA), - MyRecordB(MyRecordB), - } - - fn schema() -> Schema { - Schema::parse_str(NULLABLE_RECORD_SCHEMA).unwrap() - } - - #[test] - fn null_variant_enum_my_record_a_27() -> TestResult { - assert_roundtrip(MyUnionNullable::MyRecordA(MyRecordA { a: 27 }), &schema()) - } - - #[test] - fn rusty_my_record_a_27() { - assert_roundtrip( - Some(MyUnionAvroJsonEncoding::MyRecordA(MyRecordA { a: 27 })), - &schema(), - ) - .unwrap(); - } - - #[test] - fn rusty_untagged_my_record_a_27() { - assert_roundtrip( - Some(MyUnionUntagged::MyRecordA(MyRecordA { a: 27 })), - &schema(), - ) - .unwrap(); - } - - #[test] - fn null_variant_enum_my_record_b_27() -> TestResult { - assert_roundtrip(MyUnionNullable::MyRecordB(MyRecordB { a: 27 }), &schema()) - } - - #[test] - fn rusty_my_record_b_27() { - assert_roundtrip( - Some(MyUnionAvroJsonEncoding::MyRecordB(MyRecordB { a: 27 })), - &schema(), - ) - .unwrap(); - } - - #[test] - #[should_panic] - fn rusty_untagged_my_record_b_27() { - assert_roundtrip( - Some(MyUnionUntagged::MyRecordB(MyRecordB { a: 27 })), - &schema(), - ) - .unwrap(); - } -}