Skip to content

Support flattening Option<T> into the UnionSchema of T #528

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
PookieBuns wants to merge 6 commits into
base: main
Choose a base branch
from
Open

Support flattening Option<T> into the UnionSchema of T #528

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
038050f
feat: Support flatten Option<T> to UnionSchema
PookieBuns Mar 26, 2026
7ede683
feat: Use index bumping for flattening Option to Union Schema for ser
PookieBuns Mar 26, 2026
989b66e
feat: Use index bumping for flatten Option for deser
PookieBuns Mar 27, 2026
ffc4301
feat: Support Option flattening for AvroSchemaComponent
PookieBuns Mar 31, 2026
7c80e15
fix: Panic on nested Null unions for Option<T>
PookieBuns Apr 3, 2026
ac61167
fix: Improve documentation and simplify code
Kriskras99 Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions avro/src/documentation/serde_data_model_to_avro.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T>`.
//! - **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)
Expand Down
52 changes: 42 additions & 10 deletions avro/src/serde/derive.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -583,15 +583,27 @@ impl<T> AvroSchemaComponent for Option<T>
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<Name>,
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(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(
UnionSchema::new(variants).expect("Option<T> must produce a valid (non-nested) union"),
)
Expand Down Expand Up @@ -950,11 +962,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},
writer::datum::GenericDatumWriter,
};
use std::collections::HashSet;

#[test]
fn avro_rs_401_str() -> TestResult {
Expand Down Expand Up @@ -1070,9 +1083,7 @@ mod tests {
}

#[test]
#[should_panic(
expected = "Option<T> must produce a valid (non-nested) union: Error { details: Unions may not directly contain a union }"
)]
#[should_panic(expected = "Unions cannot contain duplicate types, found at least two Null")]
fn avro_rs_489_option_option() {
<Option<Option<i32>>>::get_schema();
}
Expand Down Expand Up @@ -1207,4 +1218,25 @@ 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(_: &mut HashSet<Name>, _: NamespaceRef) -> Schema {
Schema::union(vec![Schema::Int, Schema::String]).expect("Union must be valid")
}
}

assert_eq!(schema, Option::<MyUnion>::get_schema());

Ok(())
}
}
31 changes: 26 additions & 5 deletions avro/src/serde/deser_schema/enums.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,14 +102,31 @@ pub struct UnionEnumDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
reader: &'r mut R,
variants: &'s [Schema],
config: Config<'s, S>,
/// The index of the null that belongs to the Option<T> schema and has already been read.
flattened_option_null_index: Option<usize>,
}

impl<'s, 'r, R: Read, S: Borrow<Schema>> UnionEnumDeserializer<'s, 'r, R, S> {
pub fn new(reader: &'r mut R, schema: &'s UnionSchema, config: Config<'s, S>) -> Self {
pub fn new(
reader: &'r mut R,
schema: &'s UnionSchema,
config: Config<'s, S>,
flattened_option_null_index: Option<usize>,
) -> Self {
Self {
reader,
variants: schema.variants(),
config,
flattened_option_null_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<T> null schema,
// as Serde is not aware of the flattening.
Some(null_index) if serde_index >= null_index => serde_index - 1,
_ => serde_index,
}
}
}
Expand All @@ -124,15 +141,19 @@ impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> 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 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.correct_index_for_serde(index);
Ok((
seed.deserialize(IdentifierDeserializer::index(index as u32))?,
seed.deserialize(IdentifierDeserializer::index(variant_index as u32))?,
UnionVariantAccess::new(schema, self.reader, self.config)?,
))
}
Expand Down
36 changes: 27 additions & 9 deletions avro/src/serde/deser_schema/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Schema>> {
Expand Down Expand Up @@ -79,6 +78,7 @@ pub struct SchemaAwareDeserializer<'s, 'r, R: Read, S: Borrow<Schema>> {
/// This schema is guaranteed to not be a [`Schema::Ref`].
schema: &'s Schema,
config: Config<'s, S>,
flattened_option_null_index: Option<usize>,
}

impl<'s, 'r, R: Read, S: Borrow<Schema>> SchemaAwareDeserializer<'s, 'r, R, S> {
Expand All @@ -96,12 +96,14 @@ impl<'s, 'r, R: Read, S: Borrow<Schema>> SchemaAwareDeserializer<'s, 'r, R, S> {
reader,
schema,
config,
flattened_option_null_index: None,
})
} else {
Ok(Self {
reader,
schema,
config,
flattened_option_null_index: None,
})
}
}
Expand Down Expand Up @@ -129,12 +131,22 @@ impl<'s, 'r, R: Read, S: Borrow<Schema>> SchemaAwareDeserializer<'s, 'r, R, S> {
Ok(self)
}

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

/// 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<Self, Error> {
let index = zag_i32(self.reader)?;
let index = 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)
}
Expand Down Expand Up @@ -539,7 +551,6 @@ impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> 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)?;
Expand All @@ -548,7 +559,11 @@ impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> 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_flattened_option_null_index(index))
}
}
} else {
Err(self.error("option", "Expected Schema::Union([Schema::Null, _])"))
Expand Down Expand Up @@ -723,9 +738,12 @@ impl<'de, 's, 'r, R: Read, S: Borrow<Schema>> Deserializer<'de>
Schema::Enum(schema) => {
visitor.visit_enum(PlainEnumDeserializer::new(self.reader, schema))
}
Schema::Union(union) => {
visitor.visit_enum(UnionEnumDeserializer::new(self.reader, union, self.config))
}
Schema::Union(union) => visitor.visit_enum(UnionEnumDeserializer::new(
self.reader,
union,
self.config,
self.flattened_option_null_index,
)),
_ => Err(self.error("enum", "Expected Schema::Enum | Schema::Union")),
}
}
Expand Down
Loading