1use crate::schema::{
21 AVRO_ENUM_SYMBOLS_METADATA_KEY, AVRO_FIELD_DEFAULT_METADATA_KEY, AVRO_NAME_METADATA_KEY,
22 AVRO_NAMESPACE_METADATA_KEY, Array, Attributes, ComplexType, Enum, Fixed, Map, Nullability,
23 PrimitiveType, Record, Schema, Type, TypeName, make_full_name,
24};
25use arrow_schema::{
26 ArrowError, DECIMAL128_MAX_PRECISION, DECIMAL256_MAX_PRECISION, DataType, Field, Fields,
27 IntervalUnit, TimeUnit, UnionFields, UnionMode,
28};
29#[cfg(feature = "small_decimals")]
30use arrow_schema::{DECIMAL32_MAX_PRECISION, DECIMAL64_MAX_PRECISION};
31use indexmap::IndexMap;
32use serde_json::Value;
33use std::collections::hash_map::Entry;
34use std::collections::{HashMap, HashSet};
35use std::fmt;
36use std::fmt::Display;
37use std::sync::Arc;
38use strum_macros::AsRefStr;
39
40#[derive(Debug, Clone, PartialEq)]
42pub(crate) enum ResolutionInfo {
43 Promotion(Promotion),
45 DefaultValue(AvroLiteral),
47 EnumMapping(EnumMapping),
49 Record(ResolvedRecord),
51 Union(ResolvedUnion),
53}
54
55#[derive(Debug, Clone, PartialEq)]
59pub(crate) enum AvroLiteral {
60 Null,
62 Boolean(bool),
64 Int(i32),
66 Long(i64),
68 Float(f32),
70 Double(f64),
72 Bytes(Vec<u8>),
74 String(String),
76 Enum(String),
78 Array(Vec<AvroLiteral>),
80 Map(IndexMap<String, AvroLiteral>),
82}
83
84#[derive(Debug, Clone, PartialEq)]
86pub(crate) struct ResolvedRecord {
87 pub(crate) writer_fields: Arc<[ResolvedField]>,
89 pub(crate) default_fields: Arc<[usize]>,
91}
92
93#[derive(Debug, Clone, PartialEq)]
95pub(crate) enum ResolvedField {
96 ToReader(usize, AvroDataType),
100 Skip(AvroDataType),
103}
104
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
110pub(crate) enum Promotion {
111 Direct,
113 IntToLong,
115 IntToFloat,
117 IntToDouble,
119 LongToFloat,
121 LongToDouble,
123 FloatToDouble,
125 StringToBytes,
127 BytesToString,
129}
130
131impl Display for Promotion {
132 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
133 match self {
134 Self::Direct => write!(formatter, "Direct"),
135 Self::IntToLong => write!(formatter, "Int->Long"),
136 Self::IntToFloat => write!(formatter, "Int->Float"),
137 Self::IntToDouble => write!(formatter, "Int->Double"),
138 Self::LongToFloat => write!(formatter, "Long->Float"),
139 Self::LongToDouble => write!(formatter, "Long->Double"),
140 Self::FloatToDouble => write!(formatter, "Float->Double"),
141 Self::StringToBytes => write!(formatter, "String->Bytes"),
142 Self::BytesToString => write!(formatter, "Bytes->String"),
143 }
144 }
145}
146
147#[derive(Debug, Clone, PartialEq)]
149pub(crate) struct ResolvedUnion {
150 pub(crate) writer_to_reader: Arc<[Option<(usize, ResolutionInfo)>]>,
153 pub(crate) writer_is_union: bool,
155 pub(crate) reader_is_union: bool,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq)]
163pub(crate) struct EnumMapping {
164 pub(crate) mapping: Arc<[i32]>,
166 pub(crate) default_index: i32,
169}
170
171#[cfg(feature = "canonical_extension_types")]
172fn with_extension_type(codec: &Codec, field: Field) -> Field {
173 match codec {
174 Codec::Uuid => field.with_extension_type(arrow_schema::extension::Uuid),
175 _ => field,
176 }
177}
178
179#[derive(Debug, Clone, PartialEq)]
181pub(crate) struct AvroDataType {
182 nullability: Option<Nullability>,
183 metadata: HashMap<String, String>,
184 codec: Codec,
185 pub(crate) resolution: Option<ResolutionInfo>,
186}
187
188impl AvroDataType {
189 pub(crate) fn new(
191 codec: Codec,
192 metadata: HashMap<String, String>,
193 nullability: Option<Nullability>,
194 ) -> Self {
195 AvroDataType {
196 codec,
197 metadata,
198 nullability,
199 resolution: None,
200 }
201 }
202
203 #[inline]
204 fn new_with_resolution(
205 codec: Codec,
206 metadata: HashMap<String, String>,
207 nullability: Option<Nullability>,
208 resolution: Option<ResolutionInfo>,
209 ) -> Self {
210 Self {
211 codec,
212 metadata,
213 nullability,
214 resolution,
215 }
216 }
217
218 pub(crate) fn field_with_name(&self, name: &str) -> Field {
220 let mut nullable = self.nullability.is_some();
221 if !nullable {
222 if let Codec::Union(children, _, _) = self.codec() {
223 if children.iter().any(|c| matches!(c.codec(), Codec::Null)) {
225 nullable = true;
226 }
227 }
228 }
229 let data_type = self.codec.data_type();
230 let field = Field::new(name, data_type, nullable).with_metadata(self.metadata.clone());
231 #[cfg(feature = "canonical_extension_types")]
232 return with_extension_type(&self.codec, field);
233 #[cfg(not(feature = "canonical_extension_types"))]
234 field
235 }
236
237 pub(crate) fn codec(&self) -> &Codec {
242 &self.codec
243 }
244
245 pub(crate) fn nullability(&self) -> Option<Nullability> {
253 self.nullability
254 }
255
256 #[inline]
257 fn parse_default_literal(&self, default_json: &Value) -> Result<AvroLiteral, ArrowError> {
258 fn expect_string<'v>(
259 default_json: &'v Value,
260 data_type: &str,
261 ) -> Result<&'v str, ArrowError> {
262 match default_json {
263 Value::String(s) => Ok(s.as_str()),
264 _ => Err(ArrowError::SchemaError(format!(
265 "Default value must be a JSON string for {data_type}"
266 ))),
267 }
268 }
269
270 fn parse_bytes_default(
271 default_json: &Value,
272 expected_len: Option<usize>,
273 ) -> Result<Vec<u8>, ArrowError> {
274 let s = expect_string(default_json, "bytes/fixed logical types")?;
275 let mut out = Vec::with_capacity(s.len());
276 for ch in s.chars() {
277 let cp = ch as u32;
278 if cp > 0xFF {
279 return Err(ArrowError::SchemaError(format!(
280 "Invalid codepoint U+{cp:04X} in bytes/fixed default; must be ≤ 0xFF"
281 )));
282 }
283 out.push(cp as u8);
284 }
285 if let Some(len) = expected_len {
286 if out.len() != len {
287 return Err(ArrowError::SchemaError(format!(
288 "Default length {} does not match expected fixed size {len}",
289 out.len(),
290 )));
291 }
292 }
293 Ok(out)
294 }
295
296 fn parse_json_i64(default_json: &Value, data_type: &str) -> Result<i64, ArrowError> {
297 match default_json {
298 Value::Number(n) => n.as_i64().ok_or_else(|| {
299 ArrowError::SchemaError(format!("Default {data_type} must be an integer"))
300 }),
301 _ => Err(ArrowError::SchemaError(format!(
302 "Default {data_type} must be a JSON integer"
303 ))),
304 }
305 }
306
307 fn parse_json_f64(default_json: &Value, data_type: &str) -> Result<f64, ArrowError> {
308 match default_json {
309 Value::Number(n) => n.as_f64().ok_or_else(|| {
310 ArrowError::SchemaError(format!("Default {data_type} must be a number"))
311 }),
312 _ => Err(ArrowError::SchemaError(format!(
313 "Default {data_type} must be a JSON number"
314 ))),
315 }
316 }
317
318 if default_json.is_null() {
320 return match self.codec() {
321 Codec::Null => Ok(AvroLiteral::Null),
322 Codec::Union(encodings, _, _) if !encodings.is_empty()
323 && matches!(encodings[0].codec(), Codec::Null) =>
324 {
325 Ok(AvroLiteral::Null)
326 }
327 _ if self.nullability() == Some(Nullability::NullFirst) => Ok(AvroLiteral::Null),
328 _ => Err(ArrowError::SchemaError(
329 "JSON null default is only valid for `null` type or for a union whose first branch is `null`"
330 .to_string(),
331 )),
332 };
333 }
334 let lit = match self.codec() {
335 Codec::Null => {
336 return Err(ArrowError::SchemaError(
337 "Default for `null` type must be JSON null".to_string(),
338 ));
339 }
340 Codec::Boolean => match default_json {
341 Value::Bool(b) => AvroLiteral::Boolean(*b),
342 _ => {
343 return Err(ArrowError::SchemaError(
344 "Boolean default must be a JSON boolean".to_string(),
345 ));
346 }
347 },
348 Codec::Int32 | Codec::Date32 | Codec::TimeMillis => {
349 let i = parse_json_i64(default_json, "int")?;
350 if i < i32::MIN as i64 || i > i32::MAX as i64 {
351 return Err(ArrowError::SchemaError(format!(
352 "Default int {i} out of i32 range"
353 )));
354 }
355 AvroLiteral::Int(i as i32)
356 }
357 Codec::Int64
358 | Codec::TimeMicros
359 | Codec::TimestampMillis(_)
360 | Codec::TimestampMicros(_)
361 | Codec::TimestampNanos(_) => AvroLiteral::Long(parse_json_i64(default_json, "long")?),
362 #[cfg(feature = "avro_custom_types")]
363 Codec::DurationNanos
364 | Codec::DurationMicros
365 | Codec::DurationMillis
366 | Codec::DurationSeconds => AvroLiteral::Long(parse_json_i64(default_json, "long")?),
367 #[cfg(feature = "avro_custom_types")]
368 Codec::Int8 => {
369 let i = parse_json_i64(default_json, "int")?;
370 if i < i8::MIN as i64 || i > i8::MAX as i64 {
371 return Err(ArrowError::SchemaError(format!(
372 "Default int8 {i} out of i8 range"
373 )));
374 }
375 AvroLiteral::Int(i as i32)
376 }
377 #[cfg(feature = "avro_custom_types")]
378 Codec::Int16 => {
379 let i = parse_json_i64(default_json, "int")?;
380 if i < i16::MIN as i64 || i > i16::MAX as i64 {
381 return Err(ArrowError::SchemaError(format!(
382 "Default int16 {i} out of i16 range"
383 )));
384 }
385 AvroLiteral::Int(i as i32)
386 }
387 #[cfg(feature = "avro_custom_types")]
388 Codec::UInt8 => {
389 let i = parse_json_i64(default_json, "int")?;
390 if i < 0 || i > u8::MAX as i64 {
391 return Err(ArrowError::SchemaError(format!(
392 "Default uint8 {i} out of u8 range"
393 )));
394 }
395 AvroLiteral::Int(i as i32)
396 }
397 #[cfg(feature = "avro_custom_types")]
398 Codec::UInt16 => {
399 let i = parse_json_i64(default_json, "int")?;
400 if i < 0 || i > u16::MAX as i64 {
401 return Err(ArrowError::SchemaError(format!(
402 "Default uint16 {i} out of u16 range"
403 )));
404 }
405 AvroLiteral::Int(i as i32)
406 }
407 #[cfg(feature = "avro_custom_types")]
408 Codec::UInt32 => {
409 let i = parse_json_i64(default_json, "long")?;
410 if i < 0 || i > u32::MAX as i64 {
411 return Err(ArrowError::SchemaError(format!(
412 "Default uint32 {i} out of u32 range"
413 )));
414 }
415 AvroLiteral::Long(i)
416 }
417 #[cfg(feature = "avro_custom_types")]
418 Codec::Date64 | Codec::TimeNanos | Codec::TimestampSecs(_) => {
419 AvroLiteral::Long(parse_json_i64(default_json, "long")?)
420 }
421 #[cfg(feature = "avro_custom_types")]
422 Codec::UInt64 => AvroLiteral::Bytes(parse_bytes_default(default_json, Some(8))?),
423 #[cfg(feature = "avro_custom_types")]
424 Codec::Float16 => AvroLiteral::Bytes(parse_bytes_default(default_json, Some(2))?),
425 #[cfg(feature = "avro_custom_types")]
426 Codec::Time32Secs => {
427 let i = parse_json_i64(default_json, "int")?;
428 if i < i32::MIN as i64 || i > i32::MAX as i64 {
429 return Err(ArrowError::SchemaError(format!(
430 "Default time32-secs {i} out of i32 range"
431 )));
432 }
433 AvroLiteral::Int(i as i32)
434 }
435 #[cfg(feature = "avro_custom_types")]
436 Codec::IntervalYearMonth => {
437 AvroLiteral::Bytes(parse_bytes_default(default_json, Some(4))?)
438 }
439 #[cfg(feature = "avro_custom_types")]
440 Codec::IntervalMonthDayNano => {
441 AvroLiteral::Bytes(parse_bytes_default(default_json, Some(16))?)
442 }
443 #[cfg(feature = "avro_custom_types")]
444 Codec::IntervalDayTime => {
445 AvroLiteral::Bytes(parse_bytes_default(default_json, Some(8))?)
446 }
447 Codec::Float32 => {
448 let f = parse_json_f64(default_json, "float")?;
449 if !f.is_finite() || f < f32::MIN as f64 || f > f32::MAX as f64 {
450 return Err(ArrowError::SchemaError(format!(
451 "Default float {f} out of f32 range or not finite"
452 )));
453 }
454 AvroLiteral::Float(f as f32)
455 }
456 Codec::Float64 => AvroLiteral::Double(parse_json_f64(default_json, "double")?),
457 Codec::Utf8 | Codec::Utf8View | Codec::Uuid => {
458 AvroLiteral::String(expect_string(default_json, "string/uuid")?.to_string())
459 }
460 Codec::Binary => AvroLiteral::Bytes(parse_bytes_default(default_json, None)?),
461 Codec::Fixed(sz) => {
462 AvroLiteral::Bytes(parse_bytes_default(default_json, Some(*sz as usize))?)
463 }
464 Codec::Decimal(_, _, fixed_size) => {
465 AvroLiteral::Bytes(parse_bytes_default(default_json, *fixed_size)?)
466 }
467 Codec::Enum(symbols) => {
468 let s = expect_string(default_json, "enum")?;
469 if symbols.iter().any(|sym| sym == s) {
470 AvroLiteral::Enum(s.to_string())
471 } else {
472 return Err(ArrowError::SchemaError(format!(
473 "Default enum symbol {s:?} not found in reader enum symbols"
474 )));
475 }
476 }
477 Codec::Interval => AvroLiteral::Bytes(parse_bytes_default(default_json, Some(12))?),
478 Codec::List(item_dt) => match default_json {
479 Value::Array(items) => AvroLiteral::Array(
480 items
481 .iter()
482 .map(|v| item_dt.parse_default_literal(v))
483 .collect::<Result<_, _>>()?,
484 ),
485 _ => {
486 return Err(ArrowError::SchemaError(
487 "Default value must be a JSON array for Avro array type".to_string(),
488 ));
489 }
490 },
491 Codec::Map(val_dt) => match default_json {
492 Value::Object(map) => {
493 let mut out = IndexMap::with_capacity(map.len());
494 for (k, v) in map {
495 out.insert(k.clone(), val_dt.parse_default_literal(v)?);
496 }
497 AvroLiteral::Map(out)
498 }
499 _ => {
500 return Err(ArrowError::SchemaError(
501 "Default value must be a JSON object for Avro map type".to_string(),
502 ));
503 }
504 },
505 Codec::Struct(fields) => match default_json {
506 Value::Object(obj) => {
507 let mut out: IndexMap<String, AvroLiteral> =
508 IndexMap::with_capacity(fields.len());
509 for f in fields.as_ref() {
510 let name = f.name().to_string();
511 if let Some(sub) = obj.get(&name) {
512 out.insert(name, f.data_type().parse_default_literal(sub)?);
513 } else {
514 let stored_default =
516 f.data_type().metadata.get(AVRO_FIELD_DEFAULT_METADATA_KEY);
517 if stored_default.is_none()
518 && f.data_type().nullability() == Some(Nullability::default())
519 {
520 out.insert(name, AvroLiteral::Null);
521 } else if let Some(default_json) = stored_default {
522 let v: Value =
523 serde_json::from_str(default_json).map_err(|e| {
524 ArrowError::SchemaError(format!(
525 "Failed to parse stored subfield default JSON for '{}': {e}",
526 f.name(),
527 ))
528 })?;
529 out.insert(name, f.data_type().parse_default_literal(&v)?);
530 } else {
531 return Err(ArrowError::SchemaError(format!(
532 "Record default missing required subfield '{}' with non-nullable type {:?}",
533 f.name(),
534 f.data_type().codec()
535 )));
536 }
537 }
538 }
539 AvroLiteral::Map(out)
540 }
541 _ => {
542 return Err(ArrowError::SchemaError(
543 "Default value for record/struct must be a JSON object".to_string(),
544 ));
545 }
546 },
547 Codec::Union(encodings, _, _) => {
548 let Some(default_encoding) = encodings.first() else {
549 return Err(ArrowError::SchemaError(
550 "Union with no branches cannot have a default".to_string(),
551 ));
552 };
553 default_encoding.parse_default_literal(default_json)?
554 }
555 #[cfg(feature = "avro_custom_types")]
556 Codec::RunEndEncoded(values, _) => values.parse_default_literal(default_json)?,
557 };
558 Ok(lit)
559 }
560
561 fn store_default(&mut self, default_json: &Value) -> Result<(), ArrowError> {
562 let json_text = serde_json::to_string(default_json).map_err(|e| {
563 ArrowError::ParseError(format!("Failed to serialize default to JSON: {e}"))
564 })?;
565 self.metadata
566 .insert(AVRO_FIELD_DEFAULT_METADATA_KEY.to_string(), json_text);
567 Ok(())
568 }
569
570 fn parse_and_store_default(&mut self, default_json: &Value) -> Result<AvroLiteral, ArrowError> {
571 let lit = self.parse_default_literal(default_json)?;
572 self.store_default(default_json)?;
573 Ok(lit)
574 }
575}
576
577#[derive(Debug, Clone, PartialEq)]
579pub(crate) struct AvroField {
580 name: String,
581 data_type: AvroDataType,
582}
583
584impl AvroField {
585 pub(crate) fn field(&self) -> Field {
587 self.data_type.field_with_name(&self.name)
588 }
589
590 pub(crate) fn data_type(&self) -> &AvroDataType {
592 &self.data_type
593 }
594
595 pub(crate) fn with_utf8view(&self) -> Self {
604 let mut field = self.clone();
605 if let Codec::Utf8 = field.data_type.codec {
606 field.data_type.codec = Codec::Utf8View;
607 }
608 field
609 }
610
611 pub(crate) fn name(&self) -> &str {
616 &self.name
617 }
618}
619
620impl<'a> TryFrom<&Schema<'a>> for AvroField {
621 type Error = ArrowError;
622
623 fn try_from(schema: &Schema<'a>) -> Result<Self, Self::Error> {
624 match schema {
625 Schema::Complex(ComplexType::Record(r)) => {
626 let mut resolver = Maker::new(false, false, Tz::default());
627 let data_type = resolver.make_data_type(schema, None, None)?;
628 Ok(AvroField {
629 data_type,
630 name: r.name.to_string(),
631 })
632 }
633 _ => Err(ArrowError::ParseError(format!(
634 "Expected record got {schema:?}"
635 ))),
636 }
637 }
638}
639
640#[derive(Debug)]
642pub(crate) struct AvroFieldBuilder<'a> {
643 writer_schema: &'a Schema<'a>,
644 reader_schema: Option<&'a Schema<'a>>,
645 use_utf8view: bool,
646 strict_mode: bool,
647 tz: Tz,
648}
649
650impl<'a> AvroFieldBuilder<'a> {
651 pub(crate) fn new(writer_schema: &'a Schema<'a>) -> Self {
653 Self {
654 writer_schema,
655 reader_schema: None,
656 use_utf8view: false,
657 strict_mode: false,
658 tz: Tz::default(),
659 }
660 }
661
662 #[inline]
667 pub(crate) fn with_reader_schema(mut self, reader_schema: &'a Schema<'a>) -> Self {
668 self.reader_schema = Some(reader_schema);
669 self
670 }
671
672 pub(crate) fn with_utf8view(mut self, use_utf8view: bool) -> Self {
674 self.use_utf8view = use_utf8view;
675 self
676 }
677
678 pub(crate) fn with_strict_mode(mut self, strict_mode: bool) -> Self {
680 self.strict_mode = strict_mode;
681 self
682 }
683
684 pub(crate) fn with_tz(mut self, tz: Tz) -> Self {
686 self.tz = tz;
687 self
688 }
689
690 pub(crate) fn build(self) -> Result<AvroField, ArrowError> {
692 match self.writer_schema {
693 Schema::Complex(ComplexType::Record(r)) => {
694 let mut resolver = Maker::new(self.use_utf8view, self.strict_mode, self.tz);
695 let data_type =
696 resolver.make_data_type(self.writer_schema, self.reader_schema, None)?;
697 Ok(AvroField {
698 name: r.name.to_string(),
699 data_type,
700 })
701 }
702 _ => Err(ArrowError::ParseError(format!(
703 "Expected a Record schema to build an AvroField, but got {:?}",
704 self.writer_schema
705 ))),
706 }
707 }
708}
709
710#[derive(Debug, Copy, Clone, PartialEq, Default)]
716pub enum Tz {
717 #[default]
719 OffsetZero,
720 Utc,
722}
723
724impl Tz {
725 pub fn as_str(&self) -> &'static str {
727 match self {
728 Self::OffsetZero => "+00:00",
729 Self::Utc => "UTC",
730 }
731 }
732}
733
734impl Display for Tz {
735 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
736 f.write_str(self.as_str())
737 }
738}
739
740#[derive(Debug, Clone, PartialEq)]
744pub(crate) enum Codec {
745 Null,
747 Boolean,
749 Int32,
751 Int64,
753 Float32,
755 Float64,
757 Binary,
759 Utf8,
761 Utf8View,
766 Date32,
768 TimeMillis,
770 TimeMicros,
772 TimestampMillis(Option<Tz>),
777 TimestampMicros(Option<Tz>),
782 TimestampNanos(Option<Tz>),
787 Fixed(i32),
790 Decimal(usize, Option<usize>, Option<usize>),
797 Uuid,
799 Enum(Arc<[String]>),
803 List(Arc<AvroDataType>),
805 Struct(Arc<[AvroField]>),
807 Map(Arc<AvroDataType>),
809 Interval,
811 Union(Arc<[AvroDataType]>, UnionFields, UnionMode),
813 #[cfg(feature = "avro_custom_types")]
815 DurationNanos,
816 #[cfg(feature = "avro_custom_types")]
818 DurationMicros,
819 #[cfg(feature = "avro_custom_types")]
821 DurationMillis,
822 #[cfg(feature = "avro_custom_types")]
824 DurationSeconds,
825 #[cfg(feature = "avro_custom_types")]
826 RunEndEncoded(Arc<AvroDataType>, u8),
827 #[cfg(feature = "avro_custom_types")]
829 Int8,
830 #[cfg(feature = "avro_custom_types")]
832 Int16,
833 #[cfg(feature = "avro_custom_types")]
835 UInt8,
836 #[cfg(feature = "avro_custom_types")]
838 UInt16,
839 #[cfg(feature = "avro_custom_types")]
841 UInt32,
842 #[cfg(feature = "avro_custom_types")]
844 UInt64,
845 #[cfg(feature = "avro_custom_types")]
847 Float16,
848 #[cfg(feature = "avro_custom_types")]
850 Date64,
851 #[cfg(feature = "avro_custom_types")]
853 TimeNanos,
854 #[cfg(feature = "avro_custom_types")]
856 Time32Secs,
857 #[cfg(feature = "avro_custom_types")]
860 TimestampSecs(bool),
861 #[cfg(feature = "avro_custom_types")]
863 IntervalYearMonth,
864 #[cfg(feature = "avro_custom_types")]
866 IntervalMonthDayNano,
867 #[cfg(feature = "avro_custom_types")]
869 IntervalDayTime,
870}
871
872impl Codec {
873 fn data_type(&self) -> DataType {
874 match self {
875 Self::Null => DataType::Null,
876 Self::Boolean => DataType::Boolean,
877 Self::Int32 => DataType::Int32,
878 Self::Int64 => DataType::Int64,
879 Self::Float32 => DataType::Float32,
880 Self::Float64 => DataType::Float64,
881 Self::Binary => DataType::Binary,
882 Self::Utf8 => DataType::Utf8,
883 Self::Utf8View => DataType::Utf8View,
884 Self::Date32 => DataType::Date32,
885 Self::TimeMillis => DataType::Time32(TimeUnit::Millisecond),
886 Self::TimeMicros => DataType::Time64(TimeUnit::Microsecond),
887 Self::TimestampMillis(tz) => DataType::Timestamp(
888 TimeUnit::Millisecond,
889 tz.as_ref().map(|tz| tz.as_str().into()),
890 ),
891 Self::TimestampMicros(tz) => DataType::Timestamp(
892 TimeUnit::Microsecond,
893 tz.as_ref().map(|tz| tz.as_str().into()),
894 ),
895 Self::TimestampNanos(tz) => DataType::Timestamp(
896 TimeUnit::Nanosecond,
897 tz.as_ref().map(|tz| tz.as_str().into()),
898 ),
899 Self::Interval => DataType::Interval(IntervalUnit::MonthDayNano),
900 Self::Fixed(size) => DataType::FixedSizeBinary(*size),
901 Self::Decimal(precision, scale, _size) => {
902 let p = *precision as u8;
903 let s = scale.unwrap_or(0) as i8;
904 #[cfg(feature = "small_decimals")]
905 {
906 if *precision <= DECIMAL32_MAX_PRECISION as usize {
907 DataType::Decimal32(p, s)
908 } else if *precision <= DECIMAL64_MAX_PRECISION as usize {
909 DataType::Decimal64(p, s)
910 } else if *precision <= DECIMAL128_MAX_PRECISION as usize {
911 DataType::Decimal128(p, s)
912 } else {
913 DataType::Decimal256(p, s)
914 }
915 }
916 #[cfg(not(feature = "small_decimals"))]
917 {
918 if *precision <= DECIMAL128_MAX_PRECISION as usize {
919 DataType::Decimal128(p, s)
920 } else {
921 DataType::Decimal256(p, s)
922 }
923 }
924 }
925 Self::Uuid => DataType::FixedSizeBinary(16),
926 Self::Enum(_) => {
927 DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8))
928 }
929 Self::List(f) => {
930 DataType::List(Arc::new(f.field_with_name(Field::LIST_FIELD_DEFAULT_NAME)))
931 }
932 Self::Struct(f) => DataType::Struct(f.iter().map(|x| x.field()).collect()),
933 Self::Map(value_type) => {
934 let val_field = value_type.field_with_name(Field::MAP_VALUE_FIELD_DEFAULT_NAME);
935 DataType::Map(
936 Arc::new(Field::new(
937 Field::MAP_ENTRIES_FIELD_DEFAULT_NAME,
938 DataType::Struct(Fields::from(vec![
939 Field::new(Field::MAP_KEY_FIELD_DEFAULT_NAME, DataType::Utf8, false),
940 val_field,
941 ])),
942 false,
943 )),
944 false,
945 )
946 }
947 Self::Union(_, fields, mode) => DataType::Union(fields.clone(), *mode),
948 #[cfg(feature = "avro_custom_types")]
949 Self::DurationNanos => DataType::Duration(TimeUnit::Nanosecond),
950 #[cfg(feature = "avro_custom_types")]
951 Self::DurationMicros => DataType::Duration(TimeUnit::Microsecond),
952 #[cfg(feature = "avro_custom_types")]
953 Self::DurationMillis => DataType::Duration(TimeUnit::Millisecond),
954 #[cfg(feature = "avro_custom_types")]
955 Self::DurationSeconds => DataType::Duration(TimeUnit::Second),
956 #[cfg(feature = "avro_custom_types")]
957 Self::RunEndEncoded(values, bits) => {
958 let run_ends_dt = match *bits {
959 16 => DataType::Int16,
960 32 => DataType::Int32,
961 64 => DataType::Int64,
962 _ => unreachable!(),
963 };
964 DataType::RunEndEncoded(
965 Arc::new(Field::new("run_ends", run_ends_dt, false)),
966 Arc::new(Field::new("values", values.codec().data_type(), true)),
967 )
968 }
969 #[cfg(feature = "avro_custom_types")]
970 Self::Int8 => DataType::Int8,
971 #[cfg(feature = "avro_custom_types")]
972 Self::Int16 => DataType::Int16,
973 #[cfg(feature = "avro_custom_types")]
974 Self::UInt8 => DataType::UInt8,
975 #[cfg(feature = "avro_custom_types")]
976 Self::UInt16 => DataType::UInt16,
977 #[cfg(feature = "avro_custom_types")]
978 Self::UInt32 => DataType::UInt32,
979 #[cfg(feature = "avro_custom_types")]
980 Self::UInt64 => DataType::UInt64,
981 #[cfg(feature = "avro_custom_types")]
982 Self::Float16 => DataType::Float16,
983 #[cfg(feature = "avro_custom_types")]
984 Self::Date64 => DataType::Date64,
985 #[cfg(feature = "avro_custom_types")]
986 Self::TimeNanos => DataType::Time64(TimeUnit::Nanosecond),
987 #[cfg(feature = "avro_custom_types")]
988 Self::Time32Secs => DataType::Time32(TimeUnit::Second),
989 #[cfg(feature = "avro_custom_types")]
990 Self::TimestampSecs(is_utc) => {
991 DataType::Timestamp(TimeUnit::Second, is_utc.then(|| "+00:00".into()))
992 }
993 #[cfg(feature = "avro_custom_types")]
994 Self::IntervalYearMonth => DataType::Interval(IntervalUnit::YearMonth),
995 #[cfg(feature = "avro_custom_types")]
996 Self::IntervalMonthDayNano => DataType::Interval(IntervalUnit::MonthDayNano),
997 #[cfg(feature = "avro_custom_types")]
998 Self::IntervalDayTime => DataType::Interval(IntervalUnit::DayTime),
999 }
1000 }
1001
1002 pub(crate) fn with_utf8view(self, use_utf8view: bool) -> Self {
1008 if use_utf8view && matches!(self, Self::Utf8) {
1009 Self::Utf8View
1010 } else {
1011 self
1012 }
1013 }
1014
1015 #[inline]
1016 fn union_field_name(&self) -> String {
1017 UnionFieldKind::from(self).as_ref().to_owned()
1018 }
1019}
1020
1021impl From<PrimitiveType> for Codec {
1022 fn from(value: PrimitiveType) -> Self {
1023 match value {
1024 PrimitiveType::Null => Self::Null,
1025 PrimitiveType::Boolean => Self::Boolean,
1026 PrimitiveType::Int => Self::Int32,
1027 PrimitiveType::Long => Self::Int64,
1028 PrimitiveType::Float => Self::Float32,
1029 PrimitiveType::Double => Self::Float64,
1030 PrimitiveType::Bytes => Self::Binary,
1031 PrimitiveType::String => Self::Utf8,
1032 }
1033 }
1034}
1035
1036const fn max_precision_for_fixed_bytes(n: usize) -> Option<usize> {
1045 const MAX_P: [usize; 32] = [
1050 2, 4, 6, 9, 11, 14, 16, 18, 21, 23, 26, 28, 31, 33, 35, 38, 40, 43, 45, 47, 50, 52, 55, 57,
1051 59, 62, 64, 67, 69, 71, 74, 76,
1052 ];
1053 match n {
1054 1..=32 => Some(MAX_P[n - 1]),
1055 _ => None,
1056 }
1057}
1058
1059fn parse_decimal_attributes(
1060 attributes: &Attributes,
1061 fallback_size: Option<usize>,
1062 precision_required: bool,
1063) -> Result<(usize, usize, Option<usize>), ArrowError> {
1064 let precision = attributes
1065 .additional
1066 .get("precision")
1067 .and_then(|v| v.as_u64())
1068 .or(if precision_required { None } else { Some(10) })
1069 .ok_or_else(|| ArrowError::ParseError("Decimal requires precision".to_string()))?
1070 as usize;
1071 let scale = attributes
1072 .additional
1073 .get("scale")
1074 .and_then(|v| v.as_u64())
1075 .unwrap_or(0) as usize;
1076 let size = attributes
1077 .additional
1078 .get("size")
1079 .and_then(|v| v.as_u64())
1080 .map(|s| s as usize)
1081 .or(fallback_size);
1082 if precision == 0 {
1083 return Err(ArrowError::ParseError(
1084 "Decimal requires precision > 0".to_string(),
1085 ));
1086 }
1087 if scale > precision {
1088 return Err(ArrowError::ParseError(format!(
1089 "Decimal has invalid scale > precision: scale={scale}, precision={precision}"
1090 )));
1091 }
1092 if precision > DECIMAL256_MAX_PRECISION as usize {
1093 return Err(ArrowError::ParseError(format!(
1094 "Decimal precision {precision} exceeds maximum supported by Arrow ({})",
1095 DECIMAL256_MAX_PRECISION
1096 )));
1097 }
1098 if let Some(sz) = size {
1099 let max_p = max_precision_for_fixed_bytes(sz).ok_or_else(|| {
1100 ArrowError::ParseError(format!(
1101 "Invalid fixed size for decimal: {sz}, must be between 1 and 32 bytes"
1102 ))
1103 })?;
1104 if precision > max_p {
1105 return Err(ArrowError::ParseError(format!(
1106 "Decimal precision {precision} exceeds capacity of fixed size {sz} bytes (max {max_p})"
1107 )));
1108 }
1109 }
1110 Ok((precision, scale, size))
1111}
1112
1113#[derive(Debug, Clone, Copy, PartialEq, Eq, AsRefStr)]
1114#[strum(serialize_all = "snake_case")]
1115enum UnionFieldKind {
1116 Null,
1117 Boolean,
1118 Int,
1119 Long,
1120 Float,
1121 Double,
1122 Bytes,
1123 String,
1124 Date,
1125 TimeMillis,
1126 TimeMicros,
1127 TimestampMillisUtc,
1128 TimestampMillisLocal,
1129 TimestampMicrosUtc,
1130 TimestampMicrosLocal,
1131 TimestampNanosUtc,
1132 TimestampNanosLocal,
1133 Duration,
1134 Fixed,
1135 Decimal,
1136 Enum,
1137 Array,
1138 Record,
1139 Map,
1140 Uuid,
1141 Union,
1142}
1143
1144impl From<&Codec> for UnionFieldKind {
1145 fn from(c: &Codec) -> Self {
1146 match c {
1147 Codec::Null => Self::Null,
1148 Codec::Boolean => Self::Boolean,
1149 Codec::Int32 => Self::Int,
1150 Codec::Int64 => Self::Long,
1151 Codec::Float32 => Self::Float,
1152 Codec::Float64 => Self::Double,
1153 Codec::Binary => Self::Bytes,
1154 Codec::Utf8 | Codec::Utf8View => Self::String,
1155 Codec::Date32 => Self::Date,
1156 Codec::TimeMillis => Self::TimeMillis,
1157 Codec::TimeMicros => Self::TimeMicros,
1158 Codec::TimestampMillis(Some(Tz::OffsetZero)) => Self::TimestampMillisUtc,
1159 Codec::TimestampMillis(Some(Tz::Utc)) => Self::TimestampMillisUtc,
1160 Codec::TimestampMillis(None) => Self::TimestampMillisLocal,
1161 Codec::TimestampMicros(Some(Tz::OffsetZero)) => Self::TimestampMicrosUtc,
1162 Codec::TimestampMicros(Some(Tz::Utc)) => Self::TimestampMicrosUtc,
1163 Codec::TimestampMicros(None) => Self::TimestampMicrosLocal,
1164 Codec::TimestampNanos(Some(Tz::OffsetZero)) => Self::TimestampNanosUtc,
1165 Codec::TimestampNanos(Some(Tz::Utc)) => Self::TimestampNanosUtc,
1166 Codec::TimestampNanos(None) => Self::TimestampNanosLocal,
1167 Codec::Interval => Self::Duration,
1168 Codec::Fixed(_) => Self::Fixed,
1169 Codec::Decimal(..) => Self::Decimal,
1170 Codec::Enum(_) => Self::Enum,
1171 Codec::List(_) => Self::Array,
1172 Codec::Struct(_) => Self::Record,
1173 Codec::Map(_) => Self::Map,
1174 Codec::Uuid => Self::Uuid,
1175 Codec::Union(..) => Self::Union,
1176 #[cfg(feature = "avro_custom_types")]
1177 Codec::RunEndEncoded(values, _) => UnionFieldKind::from(values.codec()),
1178 #[cfg(feature = "avro_custom_types")]
1179 Codec::DurationNanos
1180 | Codec::DurationMicros
1181 | Codec::DurationMillis
1182 | Codec::DurationSeconds => Self::Duration,
1183 #[cfg(feature = "avro_custom_types")]
1184 Codec::Int8 | Codec::Int16 | Codec::UInt8 | Codec::UInt16 => Self::Int,
1185 #[cfg(feature = "avro_custom_types")]
1186 Codec::UInt32 | Codec::Date64 | Codec::TimeNanos | Codec::TimestampSecs(_) => {
1187 Self::Long
1188 }
1189 #[cfg(feature = "avro_custom_types")]
1190 Codec::Time32Secs => Self::TimeMillis, #[cfg(feature = "avro_custom_types")]
1192 Codec::UInt64
1193 | Codec::Float16
1194 | Codec::IntervalYearMonth
1195 | Codec::IntervalMonthDayNano
1196 | Codec::IntervalDayTime => Self::Fixed,
1197 }
1198 }
1199}
1200
1201fn union_branch_name(dt: &AvroDataType) -> String {
1202 if let Some(name) = dt.metadata.get(AVRO_NAME_METADATA_KEY) {
1203 if name.contains(".") {
1204 return name.to_string();
1206 }
1207 if let Some(ns) = dt.metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1208 return format!("{ns}.{name}");
1209 }
1210 return name.to_string();
1211 }
1212 dt.codec.union_field_name()
1213}
1214
1215fn build_union_fields(encodings: &[AvroDataType]) -> Result<UnionFields, ArrowError> {
1216 let arrow_fields: Vec<Field> = encodings
1217 .iter()
1218 .map(|encoding| encoding.field_with_name(&union_branch_name(encoding)))
1219 .collect();
1220 let type_ids: Vec<i8> = (0..arrow_fields.len()).map(|i| i as i8).collect();
1221 UnionFields::try_new(type_ids, arrow_fields)
1222}
1223
1224#[derive(Debug, Default)]
1228struct Resolver<'a> {
1229 map: HashMap<(&'a str, &'a str), AvroDataType>,
1230}
1231
1232impl<'a> Resolver<'a> {
1233 fn register(&mut self, name: &'a str, namespace: Option<&'a str>, schema: AvroDataType) {
1234 self.map.insert((namespace.unwrap_or(""), name), schema);
1235 }
1236
1237 fn resolve(&self, name: &str, namespace: Option<&'a str>) -> Result<AvroDataType, ArrowError> {
1238 let (namespace, name) = name
1239 .rsplit_once('.')
1240 .unwrap_or_else(|| (namespace.unwrap_or(""), name));
1241 self.map
1242 .get(&(namespace, name))
1243 .ok_or_else(|| ArrowError::ParseError(format!("Failed to resolve {namespace}.{name}")))
1244 .cloned()
1245 }
1246}
1247
1248fn full_name_set(name: &str, ns: Option<&str>, aliases: &[&str]) -> HashSet<String> {
1249 let mut out = HashSet::with_capacity(1 + aliases.len());
1250 let (full, _) = make_full_name(name, ns, None);
1251 out.insert(full);
1252 for a in aliases {
1253 let (fa, _) = make_full_name(a, None, ns);
1254 out.insert(fa);
1255 }
1256 out
1257}
1258
1259fn names_match(
1260 writer_name: &str,
1261 writer_namespace: Option<&str>,
1262 writer_aliases: &[&str],
1263 reader_name: &str,
1264 reader_namespace: Option<&str>,
1265 reader_aliases: &[&str],
1266) -> bool {
1267 let writer_set = full_name_set(writer_name, writer_namespace, writer_aliases);
1268 let reader_set = full_name_set(reader_name, reader_namespace, reader_aliases);
1269 !writer_set.is_disjoint(&reader_set)
1271}
1272
1273fn ensure_names_match(
1274 data_type: &str,
1275 writer_name: &str,
1276 writer_namespace: Option<&str>,
1277 writer_aliases: &[&str],
1278 reader_name: &str,
1279 reader_namespace: Option<&str>,
1280 reader_aliases: &[&str],
1281) -> Result<(), ArrowError> {
1282 if names_match(
1283 writer_name,
1284 writer_namespace,
1285 writer_aliases,
1286 reader_name,
1287 reader_namespace,
1288 reader_aliases,
1289 ) {
1290 Ok(())
1291 } else {
1292 Err(ArrowError::ParseError(format!(
1293 "{data_type} name mismatch writer={writer_name}, reader={reader_name}"
1294 )))
1295 }
1296}
1297
1298fn primitive_of(schema: &Schema) -> Option<PrimitiveType> {
1299 match schema {
1300 Schema::TypeName(TypeName::Primitive(primitive)) => Some(*primitive),
1301 Schema::Type(Type {
1302 r#type: TypeName::Primitive(primitive),
1303 ..
1304 }) => Some(*primitive),
1305 _ => None,
1306 }
1307}
1308
1309fn nullable_union_variants<'x, 'y>(
1310 variant: &'y [Schema<'x>],
1311) -> Option<(Nullability, &'y Schema<'x>)> {
1312 if variant.len() != 2 {
1313 return None;
1314 }
1315 let is_null = |schema: &Schema<'x>| {
1316 matches!(
1317 schema,
1318 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null))
1319 )
1320 };
1321 match (is_null(&variant[0]), is_null(&variant[1])) {
1322 (true, false) => Some((Nullability::NullFirst, &variant[1])),
1323 (false, true) => Some((Nullability::NullSecond, &variant[0])),
1324 _ => None,
1325 }
1326}
1327
1328#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1329enum UnionBranchKey {
1330 Named(String),
1331 Primitive(PrimitiveType),
1332 Array,
1333 Map,
1334}
1335
1336fn branch_key_of<'a>(s: &Schema<'a>, enclosing_ns: Option<&'a str>) -> Option<UnionBranchKey> {
1337 let (name, namespace) = match s {
1338 Schema::TypeName(TypeName::Primitive(p))
1339 | Schema::Type(Type {
1340 r#type: TypeName::Primitive(p),
1341 ..
1342 }) => return Some(UnionBranchKey::Primitive(*p)),
1343 Schema::TypeName(TypeName::Ref(name))
1344 | Schema::Type(Type {
1345 r#type: TypeName::Ref(name),
1346 ..
1347 }) => (name, None),
1348 Schema::Complex(ComplexType::Array(_)) => return Some(UnionBranchKey::Array),
1349 Schema::Complex(ComplexType::Map(_)) => return Some(UnionBranchKey::Map),
1350 Schema::Complex(ComplexType::Record(r)) => (&r.name, r.namespace),
1351 Schema::Complex(ComplexType::Enum(e)) => (&e.name, e.namespace),
1352 Schema::Complex(ComplexType::Fixed(f)) => (&f.name, f.namespace),
1353 Schema::Union(_) => return None,
1354 };
1355 let (full, _) = make_full_name(name, namespace, enclosing_ns);
1356 Some(UnionBranchKey::Named(full))
1357}
1358
1359fn union_first_duplicate<'a>(
1360 branches: &'a [Schema<'a>],
1361 enclosing_ns: Option<&'a str>,
1362) -> Option<String> {
1363 let mut seen = HashSet::with_capacity(branches.len());
1364 for schema in branches {
1365 if let Some(key) = branch_key_of(schema, enclosing_ns) {
1366 if !seen.insert(key.clone()) {
1367 let msg = match key {
1368 UnionBranchKey::Named(full) => format!("named type {full}"),
1369 UnionBranchKey::Primitive(p) => format!("primitive {}", p.as_ref()),
1370 UnionBranchKey::Array => "array".to_string(),
1371 UnionBranchKey::Map => "map".to_string(),
1372 };
1373 return Some(msg);
1374 }
1375 }
1376 }
1377 None
1378}
1379
1380struct Maker<'a> {
1384 resolver: Resolver<'a>,
1385 use_utf8view: bool,
1386 strict_mode: bool,
1387 tz: Tz,
1388}
1389
1390impl<'a> Maker<'a> {
1391 fn new(use_utf8view: bool, strict_mode: bool, tz: Tz) -> Self {
1392 Self {
1393 resolver: Default::default(),
1394 use_utf8view,
1395 strict_mode,
1396 tz,
1397 }
1398 }
1399
1400 #[cfg(feature = "avro_custom_types")]
1401 #[inline]
1402 fn propagate_nullability_into_ree(dt: &mut AvroDataType, nb: Nullability) {
1403 if let Codec::RunEndEncoded(values, bits) = dt.codec.clone() {
1404 let mut inner = (*values).clone();
1405 inner.nullability = Some(nb);
1406 dt.codec = Codec::RunEndEncoded(Arc::new(inner), bits);
1407 }
1408 }
1409
1410 fn make_data_type<'s>(
1411 &mut self,
1412 writer_schema: &'s Schema<'a>,
1413 reader_schema: Option<&'s Schema<'a>>,
1414 namespace: Option<&'a str>,
1415 ) -> Result<AvroDataType, ArrowError> {
1416 match reader_schema {
1417 Some(reader_schema) => self.resolve_type(writer_schema, reader_schema, namespace),
1418 None => self.parse_type(writer_schema, namespace),
1419 }
1420 }
1421
1422 fn parse_type<'s>(
1435 &mut self,
1436 schema: &'s Schema<'a>,
1437 namespace: Option<&'a str>,
1438 ) -> Result<AvroDataType, ArrowError> {
1439 match schema {
1440 Schema::TypeName(TypeName::Primitive(p)) => Ok(AvroDataType::new(
1441 Codec::from(*p).with_utf8view(self.use_utf8view),
1442 Default::default(),
1443 None,
1444 )),
1445 Schema::TypeName(TypeName::Ref(name)) => self.resolver.resolve(name, namespace),
1446 Schema::Union(f) => {
1447 let null = f
1448 .iter()
1449 .position(|x| x == &Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)));
1450 match (f.len() == 2, null) {
1451 (true, Some(0)) => {
1452 let mut field = self.parse_type(&f[1], namespace)?;
1453 field.nullability = Some(Nullability::NullFirst);
1454 #[cfg(feature = "avro_custom_types")]
1455 Self::propagate_nullability_into_ree(&mut field, Nullability::NullFirst);
1456 return Ok(field);
1457 }
1458 (true, Some(1)) => {
1459 if self.strict_mode {
1460 return Err(ArrowError::SchemaError(
1461 "Found Avro union of the form ['T','null'], which is disallowed in strict_mode"
1462 .to_string(),
1463 ));
1464 }
1465 let mut field = self.parse_type(&f[0], namespace)?;
1466 field.nullability = Some(Nullability::NullSecond);
1467 #[cfg(feature = "avro_custom_types")]
1468 Self::propagate_nullability_into_ree(&mut field, Nullability::NullSecond);
1469 return Ok(field);
1470 }
1471 _ => {}
1472 }
1473 if f.iter().any(|s| matches!(s, Schema::Union(_))) {
1475 return Err(ArrowError::SchemaError(
1476 "Avro unions may not immediately contain other unions".to_string(),
1477 ));
1478 }
1479 if let Some(dup) = union_first_duplicate(f, namespace) {
1481 return Err(ArrowError::SchemaError(format!(
1482 "Avro union contains duplicate branch type: {dup}"
1483 )));
1484 }
1485 let children: Vec<AvroDataType> = f
1487 .iter()
1488 .map(|s| self.parse_type(s, namespace))
1489 .collect::<Result<_, _>>()?;
1490 let union_fields = build_union_fields(&children)?;
1492 Ok(AvroDataType::new(
1493 Codec::Union(Arc::from(children), union_fields, UnionMode::Dense),
1494 Default::default(),
1495 None,
1496 ))
1497 }
1498 Schema::Complex(c) => match c {
1499 ComplexType::Record(r) => {
1500 let namespace = r.namespace.or(namespace);
1501 let mut metadata = r.attributes.field_metadata();
1502 let fields = r
1503 .fields
1504 .iter()
1505 .map(|field| {
1506 Ok(AvroField {
1507 name: field.name.to_string(),
1508 data_type: self.parse_type(&field.r#type, namespace)?,
1509 })
1510 })
1511 .collect::<Result<_, ArrowError>>()?;
1512 metadata.insert(AVRO_NAME_METADATA_KEY.to_string(), r.name.to_string());
1513 if let Some(ns) = namespace {
1514 metadata.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
1515 }
1516 let field = AvroDataType {
1517 nullability: None,
1518 codec: Codec::Struct(fields),
1519 metadata,
1520 resolution: None,
1521 };
1522 self.resolver.register(r.name, namespace, field.clone());
1523 Ok(field)
1524 }
1525 ComplexType::Array(a) => {
1526 let field = self.parse_type(a.items.as_ref(), namespace)?;
1527 Ok(AvroDataType {
1528 nullability: None,
1529 metadata: a.attributes.field_metadata(),
1530 codec: Codec::List(Arc::new(field)),
1531 resolution: None,
1532 })
1533 }
1534 ComplexType::Fixed(f) => {
1535 let size = f.size.try_into().map_err(|e| {
1536 ArrowError::ParseError(format!("Overflow converting size to i32: {e}"))
1537 })?;
1538 let namespace = f.namespace.or(namespace);
1539 let mut metadata = f.attributes.field_metadata();
1540 metadata.insert(AVRO_NAME_METADATA_KEY.to_string(), f.name.to_string());
1541 if let Some(ns) = namespace {
1542 metadata.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
1543 }
1544 let field = match f.attributes.logical_type {
1545 Some("decimal") => {
1546 let (precision, scale, _) =
1547 parse_decimal_attributes(&f.attributes, Some(size as usize), true)?;
1548 AvroDataType {
1549 nullability: None,
1550 metadata,
1551 codec: Codec::Decimal(precision, Some(scale), Some(size as usize)),
1552 resolution: None,
1553 }
1554 }
1555 Some("duration") => {
1556 if size != 12 {
1557 return Err(ArrowError::ParseError(format!(
1558 "Invalid fixed size for Duration: {size}, must be 12"
1559 )));
1560 };
1561 AvroDataType {
1562 nullability: None,
1563 metadata,
1564 codec: Codec::Interval,
1565 resolution: None,
1566 }
1567 }
1568 Some("uuid") => {
1569 if size != 16 {
1570 return Err(ArrowError::ParseError(format!(
1571 "Invalid fixed size for UUID: {size}, must be 16"
1572 )));
1573 }
1574 metadata.insert("logicalType".into(), "uuid".into());
1575 AvroDataType {
1576 nullability: None,
1577 metadata,
1578 codec: Codec::Fixed(size),
1579 resolution: None,
1580 }
1581 }
1582 #[cfg(feature = "avro_custom_types")]
1583 Some("arrow.uint64") if size == 8 => AvroDataType {
1584 nullability: None,
1585 metadata,
1586 codec: Codec::UInt64,
1587 resolution: None,
1588 },
1589 #[cfg(feature = "avro_custom_types")]
1590 Some("arrow.float16") if size == 2 => AvroDataType {
1591 nullability: None,
1592 metadata,
1593 codec: Codec::Float16,
1594 resolution: None,
1595 },
1596 #[cfg(feature = "avro_custom_types")]
1597 Some("arrow.interval-year-month") if size == 4 => AvroDataType {
1598 nullability: None,
1599 metadata,
1600 codec: Codec::IntervalYearMonth,
1601 resolution: None,
1602 },
1603 #[cfg(feature = "avro_custom_types")]
1604 Some("arrow.interval-month-day-nano") if size == 16 => AvroDataType {
1605 nullability: None,
1606 metadata,
1607 codec: Codec::IntervalMonthDayNano,
1608 resolution: None,
1609 },
1610 #[cfg(feature = "avro_custom_types")]
1611 Some("arrow.interval-day-time") if size == 8 => AvroDataType {
1612 nullability: None,
1613 metadata,
1614 codec: Codec::IntervalDayTime,
1615 resolution: None,
1616 },
1617 _ => AvroDataType {
1618 nullability: None,
1619 metadata,
1620 codec: Codec::Fixed(size),
1621 resolution: None,
1622 },
1623 };
1624 self.resolver.register(f.name, namespace, field.clone());
1625 Ok(field)
1626 }
1627 ComplexType::Enum(e) => {
1628 let namespace = e.namespace.or(namespace);
1629 let symbols = e
1630 .symbols
1631 .iter()
1632 .map(|s| s.to_string())
1633 .collect::<Arc<[String]>>();
1634 let mut metadata = e.attributes.field_metadata();
1635 let symbols_json = serde_json::to_string(&e.symbols).map_err(|e| {
1636 ArrowError::ParseError(format!("Failed to serialize enum symbols: {e}"))
1637 })?;
1638 metadata.insert(AVRO_ENUM_SYMBOLS_METADATA_KEY.to_string(), symbols_json);
1639 metadata.insert(AVRO_NAME_METADATA_KEY.to_string(), e.name.to_string());
1640 if let Some(ns) = namespace {
1641 metadata.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
1642 }
1643 let field = AvroDataType {
1644 nullability: None,
1645 metadata,
1646 codec: Codec::Enum(symbols),
1647 resolution: None,
1648 };
1649 self.resolver.register(e.name, namespace, field.clone());
1650 Ok(field)
1651 }
1652 ComplexType::Map(m) => {
1653 let val = self.parse_type(&m.values, namespace)?;
1654 Ok(AvroDataType {
1655 nullability: None,
1656 metadata: m.attributes.field_metadata(),
1657 codec: Codec::Map(Arc::new(val)),
1658 resolution: None,
1659 })
1660 }
1661 },
1662 Schema::Type(t) => {
1663 let mut field = self.parse_type(&Schema::TypeName(t.r#type.clone()), namespace)?;
1664 match (t.attributes.logical_type, &mut field.codec) {
1666 (Some("decimal"), c @ Codec::Binary) => {
1667 let (prec, sc, _) = parse_decimal_attributes(&t.attributes, None, false)?;
1668 *c = Codec::Decimal(prec, Some(sc), None);
1669 }
1670 (Some("date"), c @ Codec::Int32) => *c = Codec::Date32,
1671 (Some("time-millis"), c @ Codec::Int32) => *c = Codec::TimeMillis,
1672 (Some("time-micros"), c @ Codec::Int64) => *c = Codec::TimeMicros,
1673 (Some("timestamp-millis"), c @ Codec::Int64) => {
1674 *c = Codec::TimestampMillis(Some(self.tz))
1675 }
1676 (Some("timestamp-micros"), c @ Codec::Int64) => {
1677 *c = Codec::TimestampMicros(Some(self.tz))
1678 }
1679 (Some("local-timestamp-millis"), c @ Codec::Int64) => {
1680 *c = Codec::TimestampMillis(None)
1681 }
1682 (Some("local-timestamp-micros"), c @ Codec::Int64) => {
1683 *c = Codec::TimestampMicros(None)
1684 }
1685 (Some("timestamp-nanos"), c @ Codec::Int64) => {
1686 *c = Codec::TimestampNanos(Some(self.tz))
1687 }
1688 (Some("local-timestamp-nanos"), c @ Codec::Int64) => {
1689 *c = Codec::TimestampNanos(None)
1690 }
1691 (Some("uuid"), c @ Codec::Utf8) => {
1692 *c = Codec::Uuid;
1696 field.metadata.insert("logicalType".into(), "uuid".into());
1697 }
1698 #[cfg(feature = "avro_custom_types")]
1699 (Some("arrow.duration-nanos"), c @ Codec::Int64) => *c = Codec::DurationNanos,
1700 #[cfg(feature = "avro_custom_types")]
1701 (Some("arrow.duration-micros"), c @ Codec::Int64) => *c = Codec::DurationMicros,
1702 #[cfg(feature = "avro_custom_types")]
1703 (Some("arrow.duration-millis"), c @ Codec::Int64) => *c = Codec::DurationMillis,
1704 #[cfg(feature = "avro_custom_types")]
1705 (Some("arrow.duration-seconds"), c @ Codec::Int64) => {
1706 *c = Codec::DurationSeconds
1707 }
1708 #[cfg(feature = "avro_custom_types")]
1709 (Some("arrow.run-end-encoded"), _) => {
1710 let bits_u8: u8 = t
1711 .attributes
1712 .additional
1713 .get("arrow.runEndIndexBits")
1714 .and_then(|v| v.as_u64())
1715 .and_then(|n| u8::try_from(n).ok())
1716 .ok_or_else(|| ArrowError::ParseError(
1717 "arrow.run-end-encoded requires 'arrow.runEndIndexBits' (one of 16, 32, or 64)"
1718 .to_string(),
1719 ))?;
1720 if bits_u8 != 16 && bits_u8 != 32 && bits_u8 != 64 {
1721 return Err(ArrowError::ParseError(format!(
1722 "Invalid 'arrow.runEndIndexBits' value {bits_u8}; must be 16, 32, or 64"
1723 )));
1724 }
1725 let values_site = field.clone();
1727 field.codec = Codec::RunEndEncoded(Arc::new(values_site), bits_u8);
1728 }
1729 #[cfg(feature = "avro_custom_types")]
1731 (Some("arrow.int8"), c @ Codec::Int32) => *c = Codec::Int8,
1732 #[cfg(feature = "avro_custom_types")]
1733 (Some("arrow.int16"), c @ Codec::Int32) => *c = Codec::Int16,
1734 #[cfg(feature = "avro_custom_types")]
1735 (Some("arrow.uint8"), c @ Codec::Int32) => *c = Codec::UInt8,
1736 #[cfg(feature = "avro_custom_types")]
1737 (Some("arrow.uint16"), c @ Codec::Int32) => *c = Codec::UInt16,
1738 #[cfg(feature = "avro_custom_types")]
1739 (Some("arrow.uint32"), c @ Codec::Int64) => *c = Codec::UInt32,
1740 #[cfg(feature = "avro_custom_types")]
1741 (Some("arrow.uint64"), c @ Codec::Fixed(8)) => *c = Codec::UInt64,
1742 #[cfg(feature = "avro_custom_types")]
1744 (Some("arrow.float16"), c @ Codec::Fixed(2)) => *c = Codec::Float16,
1745 #[cfg(feature = "avro_custom_types")]
1747 (Some("arrow.date64"), c @ Codec::Int64) => *c = Codec::Date64,
1748 #[cfg(feature = "avro_custom_types")]
1750 (Some("arrow.time64-nanosecond"), c @ Codec::Int64) => *c = Codec::TimeNanos,
1751 #[cfg(feature = "avro_custom_types")]
1752 (Some("arrow.time32-second"), c @ Codec::Int32) => *c = Codec::Time32Secs,
1753 #[cfg(feature = "avro_custom_types")]
1754 (Some("arrow.timestamp-second"), c @ Codec::Int64) => {
1755 *c = Codec::TimestampSecs(true)
1756 }
1757 #[cfg(feature = "avro_custom_types")]
1758 (Some("arrow.local-timestamp-second"), c @ Codec::Int64) => {
1759 *c = Codec::TimestampSecs(false)
1760 }
1761 #[cfg(feature = "avro_custom_types")]
1763 (Some("arrow.interval-year-month"), c @ Codec::Fixed(4)) => {
1764 *c = Codec::IntervalYearMonth
1765 }
1766 #[cfg(feature = "avro_custom_types")]
1767 (Some("arrow.interval-month-day-nano"), c @ Codec::Fixed(16)) => {
1768 *c = Codec::IntervalMonthDayNano
1769 }
1770 #[cfg(feature = "avro_custom_types")]
1771 (Some("arrow.interval-day-time"), c @ Codec::Fixed(8)) => {
1772 *c = Codec::IntervalDayTime
1773 }
1774 (Some(logical), _) => {
1775 field.metadata.insert("logicalType".into(), logical.into());
1777 }
1778 (None, _) => {}
1779 }
1780 if matches!(field.codec, Codec::Int64) {
1781 if let Some(unit) = t
1782 .attributes
1783 .additional
1784 .get("arrowTimeUnit")
1785 .and_then(|v| v.as_str())
1786 {
1787 if unit == "nanosecond" {
1788 field.codec = Codec::TimestampNanos(Some(self.tz));
1789 }
1790 }
1791 }
1792 if !t.attributes.additional.is_empty() {
1793 for (k, v) in &t.attributes.additional {
1794 field.metadata.insert(k.to_string(), v.to_string());
1795 }
1796 }
1797 Ok(field)
1798 }
1799 }
1800 }
1801
1802 fn resolve_type<'s>(
1803 &mut self,
1804 writer_schema: &'s Schema<'a>,
1805 reader_schema: &'s Schema<'a>,
1806 namespace: Option<&'a str>,
1807 ) -> Result<AvroDataType, ArrowError> {
1808 if let (Some(write_primitive), Some(read_primitive)) =
1809 (primitive_of(writer_schema), primitive_of(reader_schema))
1810 {
1811 return self.resolve_primitives(write_primitive, read_primitive, reader_schema);
1812 }
1813 match (writer_schema, reader_schema) {
1814 (Schema::Union(writer_variants), Schema::Union(reader_variants)) => {
1815 let writer_variants = writer_variants.as_slice();
1816 let reader_variants = reader_variants.as_slice();
1817 match (
1818 nullable_union_variants(writer_variants),
1819 nullable_union_variants(reader_variants),
1820 ) {
1821 (Some((w_nb, w_nonnull)), Some((r_nb, r_nonnull))) => {
1822 let mut dt = self.resolve_type(w_nonnull, r_nonnull, namespace)?;
1823 let mut writer_to_reader = vec![None, None];
1824 writer_to_reader[w_nb.non_null_index()] = Some((
1825 r_nb.non_null_index(),
1826 dt.resolution
1827 .take()
1828 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct)),
1829 ));
1830 dt.nullability = Some(w_nb);
1831 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
1832 writer_to_reader: Arc::from(writer_to_reader),
1833 writer_is_union: true,
1834 reader_is_union: true,
1835 }));
1836 #[cfg(feature = "avro_custom_types")]
1837 Self::propagate_nullability_into_ree(&mut dt, w_nb);
1838 Ok(dt)
1839 }
1840 _ => self.resolve_unions(writer_variants, reader_variants, namespace),
1841 }
1842 }
1843 (Schema::Union(writer_variants), reader_non_union) => {
1844 let writer_to_reader: Vec<Option<(usize, ResolutionInfo)>> = writer_variants
1845 .iter()
1846 .map(|writer| {
1847 self.resolve_type(writer, reader_non_union, namespace)
1848 .ok()
1849 .map(|tmp| {
1850 let resolution = tmp
1851 .resolution
1852 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct));
1853 (0usize, resolution)
1854 })
1855 })
1856 .collect();
1857 let mut dt = self.parse_type(reader_non_union, namespace)?;
1858 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
1859 writer_to_reader: Arc::from(writer_to_reader),
1860 writer_is_union: true,
1861 reader_is_union: false,
1862 }));
1863 Ok(dt)
1864 }
1865 (writer_non_union, Schema::Union(reader_variants)) => {
1866 if let Some((nullability, non_null_branch)) =
1867 nullable_union_variants(reader_variants)
1868 {
1869 let mut dt = self.resolve_type(writer_non_union, non_null_branch, namespace)?;
1870 #[cfg(feature = "avro_custom_types")]
1871 Self::propagate_nullability_into_ree(&mut dt, nullability);
1872 dt.nullability = Some(nullability);
1873 if dt.resolution.is_none() {
1876 dt.resolution = Some(ResolutionInfo::Promotion(Promotion::Direct));
1877 }
1878 Ok(dt)
1879 } else {
1880 let Some((match_idx, mut match_dt)) =
1881 self.find_best_union_match(writer_non_union, reader_variants, namespace)
1882 else {
1883 return Err(ArrowError::SchemaError(
1884 "Writer schema does not match any reader union branch".to_string(),
1885 ));
1886 };
1887 let resolution = match_dt
1893 .resolution
1894 .take()
1895 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct));
1896 let mut match_dt = Some(match_dt);
1897 let children = reader_variants
1898 .iter()
1899 .enumerate()
1900 .map(|(idx, variant)| {
1901 if idx == match_idx {
1902 Ok(match_dt.take().unwrap())
1903 } else {
1904 self.parse_type(variant, namespace)
1905 }
1906 })
1907 .collect::<Result<Vec<_>, _>>()?;
1908 let union_fields = build_union_fields(&children)?;
1909 let mut dt = AvroDataType::new(
1910 Codec::Union(children.into(), union_fields, UnionMode::Dense),
1911 Default::default(),
1912 None,
1913 );
1914 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
1915 writer_to_reader: Arc::from(vec![Some((match_idx, resolution))]),
1916 writer_is_union: false,
1917 reader_is_union: true,
1918 }));
1919 Ok(dt)
1920 }
1921 }
1922 (
1923 Schema::Complex(ComplexType::Array(writer_array)),
1924 Schema::Complex(ComplexType::Array(reader_array)),
1925 ) => self.resolve_array(writer_array, reader_array, namespace),
1926 (
1927 Schema::Complex(ComplexType::Map(writer_map)),
1928 Schema::Complex(ComplexType::Map(reader_map)),
1929 ) => self.resolve_map(writer_map, reader_map, namespace),
1930 (
1931 Schema::Complex(ComplexType::Fixed(writer_fixed)),
1932 Schema::Complex(ComplexType::Fixed(reader_fixed)),
1933 ) => self.resolve_fixed(writer_fixed, reader_fixed, reader_schema, namespace),
1934 (
1935 Schema::Complex(ComplexType::Record(writer_record)),
1936 Schema::Complex(ComplexType::Record(reader_record)),
1937 ) => self.resolve_records(writer_record, reader_record, namespace),
1938 (
1939 Schema::Complex(ComplexType::Enum(writer_enum)),
1940 Schema::Complex(ComplexType::Enum(reader_enum)),
1941 ) => self.resolve_enums(writer_enum, reader_enum, reader_schema, namespace),
1942 (Schema::TypeName(TypeName::Ref(_)), _) => self.parse_type(reader_schema, namespace),
1943 (_, Schema::TypeName(TypeName::Ref(_))) => self.parse_type(reader_schema, namespace),
1944 _ => Err(ArrowError::NotYetImplemented(
1945 "Other resolutions not yet implemented".to_string(),
1946 )),
1947 }
1948 }
1949
1950 fn find_best_union_match(
1951 &mut self,
1952 writer: &Schema<'a>,
1953 reader_variants: &[Schema<'a>],
1954 namespace: Option<&'a str>,
1955 ) -> Option<(usize, AvroDataType)> {
1956 let mut first_resolution = None;
1957 for (reader_index, reader) in reader_variants.iter().enumerate() {
1958 if let Ok(dt) = self.resolve_type(writer, reader, namespace) {
1959 match &dt.resolution {
1960 None | Some(ResolutionInfo::Promotion(Promotion::Direct)) => {
1961 return Some((reader_index, dt));
1963 }
1964 Some(_) => {
1965 if first_resolution.is_none() {
1966 first_resolution = Some((reader_index, dt));
1968 }
1969 }
1970 };
1971 }
1972 }
1973 first_resolution
1974 }
1975
1976 fn resolve_unions<'s>(
1977 &mut self,
1978 writer_variants: &'s [Schema<'a>],
1979 reader_variants: &'s [Schema<'a>],
1980 namespace: Option<&'a str>,
1981 ) -> Result<AvroDataType, ArrowError> {
1982 let mut resolved_reader_encodings = HashMap::new();
1983 let writer_to_reader: Vec<Option<(usize, ResolutionInfo)>> = writer_variants
1984 .iter()
1985 .map(|writer| {
1986 self.find_best_union_match(writer, reader_variants, namespace)
1987 .map(|(match_idx, mut match_dt)| {
1988 let resolution = match_dt
1989 .resolution
1990 .take()
1991 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct));
1992 resolved_reader_encodings.insert(match_idx, match_dt);
1995 (match_idx, resolution)
1996 })
1997 })
1998 .collect();
1999 let reader_encodings: Vec<AvroDataType> = reader_variants
2000 .iter()
2001 .enumerate()
2002 .map(|(reader_idx, reader_schema)| {
2003 if let Some(resolved) = resolved_reader_encodings.remove(&reader_idx) {
2004 Ok(resolved)
2005 } else {
2006 self.parse_type(reader_schema, namespace)
2007 }
2008 })
2009 .collect::<Result<_, _>>()?;
2010 let union_fields = build_union_fields(&reader_encodings)?;
2011 let mut dt = AvroDataType::new(
2012 Codec::Union(reader_encodings.into(), union_fields, UnionMode::Dense),
2013 Default::default(),
2014 None,
2015 );
2016 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
2017 writer_to_reader: Arc::from(writer_to_reader),
2018 writer_is_union: true,
2019 reader_is_union: true,
2020 }));
2021 Ok(dt)
2022 }
2023
2024 fn resolve_array(
2025 &mut self,
2026 writer_array: &Array<'a>,
2027 reader_array: &Array<'a>,
2028 namespace: Option<&'a str>,
2029 ) -> Result<AvroDataType, ArrowError> {
2030 Ok(AvroDataType {
2031 nullability: None,
2032 metadata: reader_array.attributes.field_metadata(),
2033 codec: Codec::List(Arc::new(self.make_data_type(
2034 writer_array.items.as_ref(),
2035 Some(reader_array.items.as_ref()),
2036 namespace,
2037 )?)),
2038 resolution: None,
2039 })
2040 }
2041
2042 fn resolve_map(
2043 &mut self,
2044 writer_map: &Map<'a>,
2045 reader_map: &Map<'a>,
2046 namespace: Option<&'a str>,
2047 ) -> Result<AvroDataType, ArrowError> {
2048 Ok(AvroDataType {
2049 nullability: None,
2050 metadata: reader_map.attributes.field_metadata(),
2051 codec: Codec::Map(Arc::new(self.make_data_type(
2052 &writer_map.values,
2053 Some(&reader_map.values),
2054 namespace,
2055 )?)),
2056 resolution: None,
2057 })
2058 }
2059
2060 fn resolve_fixed<'s>(
2061 &mut self,
2062 writer_fixed: &Fixed<'a>,
2063 reader_fixed: &Fixed<'a>,
2064 reader_schema: &'s Schema<'a>,
2065 namespace: Option<&'a str>,
2066 ) -> Result<AvroDataType, ArrowError> {
2067 ensure_names_match(
2068 "Fixed",
2069 writer_fixed.name,
2070 writer_fixed.namespace,
2071 &writer_fixed.aliases,
2072 reader_fixed.name,
2073 reader_fixed.namespace,
2074 &reader_fixed.aliases,
2075 )?;
2076 if writer_fixed.size != reader_fixed.size {
2077 return Err(ArrowError::SchemaError(format!(
2078 "Fixed size mismatch for {}: writer={}, reader={}",
2079 reader_fixed.name, writer_fixed.size, reader_fixed.size
2080 )));
2081 }
2082 self.parse_type(reader_schema, namespace)
2083 }
2084
2085 fn resolve_primitives(
2086 &mut self,
2087 write_primitive: PrimitiveType,
2088 read_primitive: PrimitiveType,
2089 reader_schema: &Schema<'a>,
2090 ) -> Result<AvroDataType, ArrowError> {
2091 if write_primitive == read_primitive {
2092 return self.parse_type(reader_schema, None);
2093 }
2094 let promotion = match (write_primitive, read_primitive) {
2095 (PrimitiveType::Int, PrimitiveType::Long) => Promotion::IntToLong,
2096 (PrimitiveType::Int, PrimitiveType::Float) => Promotion::IntToFloat,
2097 (PrimitiveType::Int, PrimitiveType::Double) => Promotion::IntToDouble,
2098 (PrimitiveType::Long, PrimitiveType::Float) => Promotion::LongToFloat,
2099 (PrimitiveType::Long, PrimitiveType::Double) => Promotion::LongToDouble,
2100 (PrimitiveType::Float, PrimitiveType::Double) => Promotion::FloatToDouble,
2101 (PrimitiveType::String, PrimitiveType::Bytes) => Promotion::StringToBytes,
2102 (PrimitiveType::Bytes, PrimitiveType::String) => Promotion::BytesToString,
2103 _ => {
2104 return Err(ArrowError::ParseError(format!(
2105 "Illegal promotion {write_primitive:?} to {read_primitive:?}"
2106 )));
2107 }
2108 };
2109 let mut datatype = self.parse_type(reader_schema, None)?;
2110 datatype.resolution = Some(ResolutionInfo::Promotion(promotion));
2111 Ok(datatype)
2112 }
2113
2114 fn resolve_enums(
2170 &mut self,
2171 writer_enum: &Enum<'a>,
2172 reader_enum: &Enum<'a>,
2173 reader_schema: &Schema<'a>,
2174 namespace: Option<&'a str>,
2175 ) -> Result<AvroDataType, ArrowError> {
2176 ensure_names_match(
2177 "Enum",
2178 writer_enum.name,
2179 writer_enum.namespace,
2180 &writer_enum.aliases,
2181 reader_enum.name,
2182 reader_enum.namespace,
2183 &reader_enum.aliases,
2184 )?;
2185 if writer_enum.symbols == reader_enum.symbols {
2186 return self.parse_type(reader_schema, namespace);
2187 }
2188 let reader_index: HashMap<&str, i32> = reader_enum
2189 .symbols
2190 .iter()
2191 .enumerate()
2192 .map(|(index, &symbol)| (symbol, index as i32))
2193 .collect();
2194 let default_index: i32 = match reader_enum.default {
2195 Some(symbol) => *reader_index.get(symbol).ok_or_else(|| {
2196 ArrowError::SchemaError(format!(
2197 "Reader enum '{}' default symbol '{symbol}' not found in symbols list",
2198 reader_enum.name,
2199 ))
2200 })?,
2201 None => -1,
2202 };
2203 let mapping: Vec<i32> = writer_enum
2204 .symbols
2205 .iter()
2206 .map(|&write_symbol| {
2207 reader_index
2208 .get(write_symbol)
2209 .copied()
2210 .unwrap_or(default_index)
2211 })
2212 .collect();
2213 if self.strict_mode && mapping.iter().any(|&m| m < 0) {
2214 return Err(ArrowError::SchemaError(format!(
2215 "Reader enum '{}' does not cover all writer symbols and no default is provided",
2216 reader_enum.name
2217 )));
2218 }
2219 let mut dt = self.parse_type(reader_schema, namespace)?;
2220 dt.resolution = Some(ResolutionInfo::EnumMapping(EnumMapping {
2221 mapping: Arc::from(mapping),
2222 default_index,
2223 }));
2224 let reader_ns = reader_enum.namespace.or(namespace);
2225 self.resolver
2226 .register(reader_enum.name, reader_ns, dt.clone());
2227 Ok(dt)
2228 }
2229
2230 #[inline]
2231 fn build_writer_lookup(
2232 writer_record: &Record<'a>,
2233 ) -> (HashMap<&'a str, usize>, HashSet<&'a str>) {
2234 let mut map: HashMap<&str, usize> = HashMap::with_capacity(writer_record.fields.len() * 2);
2235 for (idx, wf) in writer_record.fields.iter().enumerate() {
2236 map.insert(wf.name, idx);
2238 }
2239 let mut ambiguous: HashSet<&str> = HashSet::new();
2241 for (idx, wf) in writer_record.fields.iter().enumerate() {
2242 for &alias in &wf.aliases {
2243 match map.entry(alias) {
2244 Entry::Occupied(e) if *e.get() != idx => {
2245 ambiguous.insert(alias);
2246 }
2247 Entry::Vacant(e) => {
2248 e.insert(idx);
2249 }
2250 _ => {}
2251 }
2252 }
2253 }
2254 (map, ambiguous)
2255 }
2256
2257 fn resolve_records(
2258 &mut self,
2259 writer_record: &Record<'a>,
2260 reader_record: &Record<'a>,
2261 namespace: Option<&'a str>,
2262 ) -> Result<AvroDataType, ArrowError> {
2263 ensure_names_match(
2264 "Record",
2265 writer_record.name,
2266 writer_record.namespace,
2267 &writer_record.aliases,
2268 reader_record.name,
2269 reader_record.namespace,
2270 &reader_record.aliases,
2271 )?;
2272 let writer_ns = writer_record.namespace.or(namespace);
2273 let reader_ns = reader_record.namespace.or(namespace);
2274 let mut reader_md = reader_record.attributes.field_metadata();
2275 reader_md.insert(
2276 AVRO_NAME_METADATA_KEY.to_string(),
2277 reader_record.name.to_string(),
2278 );
2279 if let Some(ns) = reader_ns {
2280 reader_md.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
2281 }
2282 let (writer_lookup, ambiguous_writer_aliases) = Self::build_writer_lookup(writer_record);
2284 let mut writer_to_reader: Vec<Option<usize>> = vec![None; writer_record.fields.len()];
2285 let mut reader_fields: Vec<AvroField> = Vec::with_capacity(reader_record.fields.len());
2286 let mut default_fields: Vec<usize> = Vec::new();
2288 for (reader_idx, r_field) in reader_record.fields.iter().enumerate() {
2289 let mut match_idx = writer_lookup.get(r_field.name).copied();
2291 let mut matched_via_alias: Option<&str> = None;
2292 if match_idx.is_none() {
2293 for &alias in &r_field.aliases {
2294 if let Some(i) = writer_lookup.get(alias).copied() {
2295 if self.strict_mode && ambiguous_writer_aliases.contains(alias) {
2296 return Err(ArrowError::SchemaError(format!(
2297 "Ambiguous alias '{alias}' on reader field '{}' matches multiple writer fields",
2298 r_field.name
2299 )));
2300 }
2301 match_idx = Some(i);
2302 matched_via_alias = Some(alias);
2303 break;
2304 }
2305 }
2306 }
2307 if let Some(wi) = match_idx {
2308 if writer_to_reader[wi].is_none() {
2309 let w_schema = &writer_record.fields[wi].r#type;
2310 let dt = self.make_data_type(w_schema, Some(&r_field.r#type), reader_ns)?;
2311 writer_to_reader[wi] = Some(reader_idx);
2312 reader_fields.push(AvroField {
2313 name: r_field.name.to_owned(),
2314 data_type: dt,
2315 });
2316 continue;
2317 } else if self.strict_mode {
2318 let existing_reader = writer_to_reader[wi].unwrap();
2320 let via = matched_via_alias
2321 .map(|a| format!("alias '{a}'"))
2322 .unwrap_or_else(|| "name match".to_string());
2323 return Err(ArrowError::SchemaError(format!(
2324 "Multiple reader fields map to the same writer field '{}' via {via} (existing reader index {existing_reader}, new reader index {reader_idx})",
2325 writer_record.fields[wi].name
2326 )));
2327 }
2328 }
2330 let mut dt = self.parse_type(&r_field.r#type, reader_ns)?;
2332 if let Some(default_json) = r_field.default.as_ref() {
2333 dt.resolution = Some(ResolutionInfo::DefaultValue(
2334 dt.parse_and_store_default(default_json)?,
2335 ));
2336 default_fields.push(reader_idx);
2337 } else if dt.nullability() == Some(Nullability::NullFirst) {
2338 dt.resolution = Some(ResolutionInfo::DefaultValue(
2340 dt.parse_and_store_default(&Value::Null)?,
2341 ));
2342 default_fields.push(reader_idx);
2343 } else {
2344 return Err(ArrowError::SchemaError(format!(
2345 "Reader field '{}' not present in writer schema must have a default value",
2346 r_field.name
2347 )));
2348 }
2349 reader_fields.push(AvroField {
2350 name: r_field.name.to_owned(),
2351 data_type: dt,
2352 });
2353 }
2354 let writer_fields = writer_record
2356 .fields
2357 .iter()
2358 .enumerate()
2359 .map(|(writer_index, writer_field)| {
2360 let dt = self.parse_type(&writer_field.r#type, writer_ns)?;
2361 if let Some(reader_index) = writer_to_reader[writer_index] {
2362 Ok(ResolvedField::ToReader(reader_index, dt))
2363 } else {
2364 Ok(ResolvedField::Skip(dt))
2365 }
2366 })
2367 .collect::<Result<_, ArrowError>>()?;
2368 let resolved = AvroDataType::new_with_resolution(
2369 Codec::Struct(Arc::from(reader_fields)),
2370 reader_md,
2371 None,
2372 Some(ResolutionInfo::Record(ResolvedRecord {
2373 writer_fields,
2374 default_fields: Arc::from(default_fields),
2375 })),
2376 );
2377 self.resolver
2379 .register(reader_record.name, reader_ns, resolved.clone());
2380 Ok(resolved)
2381 }
2382}
2383
2384#[cfg(test)]
2385mod tests {
2386 use super::*;
2387 use crate::schema::{
2388 AVRO_ROOT_RECORD_DEFAULT_NAME, Array, Attributes, ComplexType, Field as AvroFieldSchema,
2389 Fixed, PrimitiveType, Record, Schema, Type, TypeName,
2390 };
2391 use indexmap::IndexMap;
2392 use serde_json::{self, Value};
2393
2394 fn create_schema_with_logical_type(
2395 primitive_type: PrimitiveType,
2396 logical_type: &'static str,
2397 ) -> Schema<'static> {
2398 let attributes = Attributes {
2399 logical_type: Some(logical_type),
2400 additional: Default::default(),
2401 };
2402
2403 Schema::Type(Type {
2404 r#type: TypeName::Primitive(primitive_type),
2405 attributes,
2406 })
2407 }
2408
2409 fn resolve_promotion(writer: PrimitiveType, reader: PrimitiveType) -> AvroDataType {
2410 let writer_schema = Schema::TypeName(TypeName::Primitive(writer));
2411 let reader_schema = Schema::TypeName(TypeName::Primitive(reader));
2412 let mut maker = Maker::new(false, false, Tz::default());
2413 maker
2414 .make_data_type(&writer_schema, Some(&reader_schema), None)
2415 .expect("promotion should resolve")
2416 }
2417
2418 fn mk_primitive(pt: PrimitiveType) -> Schema<'static> {
2419 Schema::TypeName(TypeName::Primitive(pt))
2420 }
2421 fn mk_union(branches: Vec<Schema<'_>>) -> Schema<'_> {
2422 Schema::Union(branches)
2423 }
2424
2425 #[test]
2426 fn test_date_logical_type() {
2427 let schema = create_schema_with_logical_type(PrimitiveType::Int, "date");
2428
2429 let mut maker = Maker::new(false, false, Tz::default());
2430 let result = maker.make_data_type(&schema, None, None).unwrap();
2431
2432 assert!(matches!(result.codec, Codec::Date32));
2433 }
2434
2435 #[test]
2436 fn test_time_millis_logical_type() {
2437 let schema = create_schema_with_logical_type(PrimitiveType::Int, "time-millis");
2438
2439 let mut maker = Maker::new(false, false, Tz::default());
2440 let result = maker.make_data_type(&schema, None, None).unwrap();
2441
2442 assert!(matches!(result.codec, Codec::TimeMillis));
2443 }
2444
2445 #[test]
2446 fn test_time_micros_logical_type() {
2447 let schema = create_schema_with_logical_type(PrimitiveType::Long, "time-micros");
2448
2449 let mut maker = Maker::new(false, false, Tz::default());
2450 let result = maker.make_data_type(&schema, None, None).unwrap();
2451
2452 assert!(matches!(result.codec, Codec::TimeMicros));
2453 }
2454
2455 #[test]
2456 fn test_timestamp_millis_logical_type() {
2457 for tz in [Tz::OffsetZero, Tz::Utc] {
2458 let schema = create_schema_with_logical_type(PrimitiveType::Long, "timestamp-millis");
2459
2460 let mut maker = Maker::new(false, false, tz);
2461 let result = maker.make_data_type(&schema, None, None).unwrap();
2462
2463 let Codec::TimestampMillis(Some(actual_tz)) = result.codec else {
2464 panic!("Expected TimestampMillis codec");
2465 };
2466 assert_eq!(actual_tz, tz);
2467 }
2468 }
2469
2470 #[test]
2471 fn test_timestamp_micros_logical_type() {
2472 for tz in [Tz::OffsetZero, Tz::Utc] {
2473 let schema = create_schema_with_logical_type(PrimitiveType::Long, "timestamp-micros");
2474
2475 let mut maker = Maker::new(false, false, tz);
2476 let result = maker.make_data_type(&schema, None, None).unwrap();
2477
2478 let Codec::TimestampMicros(Some(actual_tz)) = result.codec else {
2479 panic!("Expected TimestampMicros codec");
2480 };
2481 assert_eq!(actual_tz, tz);
2482 }
2483 }
2484
2485 #[test]
2486 fn test_timestamp_nanos_logical_type() {
2487 for tz in [Tz::OffsetZero, Tz::Utc] {
2488 let schema = create_schema_with_logical_type(PrimitiveType::Long, "timestamp-nanos");
2489
2490 let mut maker = Maker::new(false, false, tz);
2491 let result = maker.make_data_type(&schema, None, None).unwrap();
2492
2493 let Codec::TimestampNanos(Some(actual_tz)) = result.codec else {
2494 panic!("Expected TimestampNanos codec");
2495 };
2496 assert_eq!(actual_tz, tz);
2497 }
2498 }
2499
2500 #[test]
2501 fn test_local_timestamp_millis_logical_type() {
2502 let schema = create_schema_with_logical_type(PrimitiveType::Long, "local-timestamp-millis");
2503
2504 let mut maker = Maker::new(false, false, Tz::default());
2505 let result = maker.make_data_type(&schema, None, None).unwrap();
2506
2507 assert!(matches!(result.codec, Codec::TimestampMillis(None)));
2508 }
2509
2510 #[test]
2511 fn test_local_timestamp_micros_logical_type() {
2512 let schema = create_schema_with_logical_type(PrimitiveType::Long, "local-timestamp-micros");
2513
2514 let mut maker = Maker::new(false, false, Tz::default());
2515 let result = maker.make_data_type(&schema, None, None).unwrap();
2516
2517 assert!(matches!(result.codec, Codec::TimestampMicros(None)));
2518 }
2519
2520 #[test]
2521 fn test_local_timestamp_nanos_logical_type() {
2522 let schema = create_schema_with_logical_type(PrimitiveType::Long, "local-timestamp-nanos");
2523
2524 let mut maker = Maker::new(false, false, Tz::default());
2525 let result = maker.make_data_type(&schema, None, None).unwrap();
2526
2527 assert!(matches!(result.codec, Codec::TimestampNanos(None)));
2528 }
2529
2530 #[test]
2531 fn test_uuid_type() {
2532 let mut codec = Codec::Fixed(16);
2533 if let c @ Codec::Fixed(16) = &mut codec {
2534 *c = Codec::Uuid;
2535 }
2536 assert!(matches!(codec, Codec::Uuid));
2537 }
2538
2539 #[test]
2540 fn test_fixed_uuid_logical_type_metadata() {
2541 let schema = Schema::Complex(ComplexType::Fixed(Fixed {
2545 name: "uuid_fixed",
2546 namespace: None,
2547 aliases: vec![],
2548 size: 16,
2549 attributes: Attributes {
2550 logical_type: Some("uuid"),
2551 additional: Default::default(),
2552 },
2553 }));
2554
2555 let mut maker = Maker::new(false, false, Tz::default());
2556 let result = maker.make_data_type(&schema, None, None).unwrap();
2557
2558 assert!(
2559 matches!(result.codec, Codec::Fixed(16)),
2560 "codec should be Fixed(16), got {:?}",
2561 result.codec
2562 );
2563 assert_eq!(
2564 result.metadata.get("logicalType").map(|s| s.as_str()),
2565 Some("uuid"),
2566 "logicalType metadata should be 'uuid'"
2567 );
2568 }
2569
2570 #[test]
2571 fn test_duration_logical_type() {
2572 let mut codec = Codec::Fixed(12);
2573
2574 if let c @ Codec::Fixed(12) = &mut codec {
2575 *c = Codec::Interval;
2576 }
2577
2578 assert!(matches!(codec, Codec::Interval));
2579 }
2580
2581 #[test]
2582 fn test_decimal_logical_type_not_implemented() {
2583 let codec = Codec::Fixed(16);
2584
2585 let process_decimal = || -> Result<(), ArrowError> {
2586 if let Codec::Fixed(_) = codec {
2587 return Err(ArrowError::NotYetImplemented(
2588 "Decimals are not currently supported".to_string(),
2589 ));
2590 }
2591 Ok(())
2592 };
2593
2594 let result = process_decimal();
2595
2596 assert!(result.is_err());
2597 if let Err(ArrowError::NotYetImplemented(msg)) = result {
2598 assert!(msg.contains("Decimals are not currently supported"));
2599 } else {
2600 panic!("Expected NotYetImplemented error");
2601 }
2602 }
2603 #[test]
2604 fn test_unknown_logical_type_added_to_metadata() {
2605 let schema = create_schema_with_logical_type(PrimitiveType::Int, "custom-type");
2606
2607 let mut maker = Maker::new(false, false, Tz::default());
2608 let result = maker.make_data_type(&schema, None, None).unwrap();
2609
2610 assert_eq!(
2611 result.metadata.get("logicalType"),
2612 Some(&"custom-type".to_string())
2613 );
2614 }
2615
2616 #[test]
2617 fn test_string_with_utf8view_enabled() {
2618 let schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
2619
2620 let mut maker = Maker::new(true, false, Tz::default());
2621 let result = maker.make_data_type(&schema, None, None).unwrap();
2622
2623 assert!(matches!(result.codec, Codec::Utf8View));
2624 }
2625
2626 #[test]
2627 fn test_string_without_utf8view_enabled() {
2628 let schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
2629
2630 let mut maker = Maker::new(false, false, Tz::default());
2631 let result = maker.make_data_type(&schema, None, None).unwrap();
2632
2633 assert!(matches!(result.codec, Codec::Utf8));
2634 }
2635
2636 #[test]
2637 fn test_record_with_string_and_utf8view_enabled() {
2638 let field_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
2639
2640 let avro_field = crate::schema::Field {
2641 name: "string_field",
2642 r#type: field_schema,
2643 default: None,
2644 doc: None,
2645 aliases: vec![],
2646 };
2647
2648 let record = Record {
2649 name: "test_record",
2650 namespace: None,
2651 aliases: vec![],
2652 doc: None,
2653 fields: vec![avro_field],
2654 attributes: Attributes::default(),
2655 };
2656
2657 let schema = Schema::Complex(ComplexType::Record(record));
2658
2659 let mut maker = Maker::new(true, false, Tz::default());
2660 let result = maker.make_data_type(&schema, None, None).unwrap();
2661
2662 if let Codec::Struct(fields) = &result.codec {
2663 let first_field_codec = &fields[0].data_type().codec;
2664 assert!(matches!(first_field_codec, Codec::Utf8View));
2665 } else {
2666 panic!("Expected Struct codec");
2667 }
2668 }
2669
2670 #[test]
2671 fn test_union_with_strict_mode() {
2672 let schema = Schema::Union(vec![
2673 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2674 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2675 ]);
2676
2677 let mut maker = Maker::new(false, true, Tz::default());
2678 let result = maker.make_data_type(&schema, None, None);
2679
2680 assert!(result.is_err());
2681 match result {
2682 Err(ArrowError::SchemaError(msg)) => {
2683 assert!(msg.contains(
2684 "Found Avro union of the form ['T','null'], which is disallowed in strict_mode"
2685 ));
2686 }
2687 _ => panic!("Expected SchemaError"),
2688 }
2689 }
2690
2691 #[test]
2692 fn test_resolve_int_to_float_promotion() {
2693 let result = resolve_promotion(PrimitiveType::Int, PrimitiveType::Float);
2694 assert!(matches!(result.codec, Codec::Float32));
2695 assert_eq!(
2696 result.resolution,
2697 Some(ResolutionInfo::Promotion(Promotion::IntToFloat))
2698 );
2699 }
2700
2701 #[test]
2702 fn test_resolve_int_to_double_promotion() {
2703 let result = resolve_promotion(PrimitiveType::Int, PrimitiveType::Double);
2704 assert!(matches!(result.codec, Codec::Float64));
2705 assert_eq!(
2706 result.resolution,
2707 Some(ResolutionInfo::Promotion(Promotion::IntToDouble))
2708 );
2709 }
2710
2711 #[test]
2712 fn test_resolve_long_to_float_promotion() {
2713 let result = resolve_promotion(PrimitiveType::Long, PrimitiveType::Float);
2714 assert!(matches!(result.codec, Codec::Float32));
2715 assert_eq!(
2716 result.resolution,
2717 Some(ResolutionInfo::Promotion(Promotion::LongToFloat))
2718 );
2719 }
2720
2721 #[test]
2722 fn test_resolve_long_to_double_promotion() {
2723 let result = resolve_promotion(PrimitiveType::Long, PrimitiveType::Double);
2724 assert!(matches!(result.codec, Codec::Float64));
2725 assert_eq!(
2726 result.resolution,
2727 Some(ResolutionInfo::Promotion(Promotion::LongToDouble))
2728 );
2729 }
2730
2731 #[test]
2732 fn test_resolve_float_to_double_promotion() {
2733 let result = resolve_promotion(PrimitiveType::Float, PrimitiveType::Double);
2734 assert!(matches!(result.codec, Codec::Float64));
2735 assert_eq!(
2736 result.resolution,
2737 Some(ResolutionInfo::Promotion(Promotion::FloatToDouble))
2738 );
2739 }
2740
2741 #[test]
2742 fn test_resolve_string_to_bytes_promotion() {
2743 let result = resolve_promotion(PrimitiveType::String, PrimitiveType::Bytes);
2744 assert!(matches!(result.codec, Codec::Binary));
2745 assert_eq!(
2746 result.resolution,
2747 Some(ResolutionInfo::Promotion(Promotion::StringToBytes))
2748 );
2749 }
2750
2751 #[test]
2752 fn test_resolve_bytes_to_string_promotion() {
2753 let result = resolve_promotion(PrimitiveType::Bytes, PrimitiveType::String);
2754 assert!(matches!(result.codec, Codec::Utf8));
2755 assert_eq!(
2756 result.resolution,
2757 Some(ResolutionInfo::Promotion(Promotion::BytesToString))
2758 );
2759 }
2760
2761 #[test]
2762 fn test_resolve_illegal_promotion_double_to_float_errors() {
2763 let writer_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Double));
2764 let reader_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Float));
2765 let mut maker = Maker::new(false, false, Tz::default());
2766 let result = maker.make_data_type(&writer_schema, Some(&reader_schema), None);
2767 assert!(result.is_err());
2768 match result {
2769 Err(ArrowError::ParseError(msg)) => {
2770 assert!(msg.contains("Illegal promotion"));
2771 }
2772 _ => panic!("Expected ParseError for illegal promotion Double -> Float"),
2773 }
2774 }
2775
2776 #[test]
2777 fn test_promotion_within_nullable_union_keeps_writer_null_ordering() {
2778 let writer = Schema::Union(vec![
2779 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2780 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2781 ]);
2782 let reader = Schema::Union(vec![
2783 Schema::TypeName(TypeName::Primitive(PrimitiveType::Double)),
2784 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2785 ]);
2786 let mut maker = Maker::new(false, false, Tz::default());
2787 let result = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2788 assert!(matches!(result.codec, Codec::Float64));
2789 assert_eq!(
2790 result.resolution,
2791 Some(ResolutionInfo::Union(ResolvedUnion {
2792 writer_to_reader: [
2793 None,
2794 Some((0, ResolutionInfo::Promotion(Promotion::IntToDouble)))
2795 ]
2796 .into(),
2797 writer_is_union: true,
2798 reader_is_union: true,
2799 }))
2800 );
2801 assert_eq!(result.nullability, Some(Nullability::NullFirst));
2802 }
2803
2804 #[test]
2805 fn test_resolve_writer_union_to_reader_non_union_partial_coverage() {
2806 let writer = mk_union(vec![
2807 mk_primitive(PrimitiveType::String),
2808 mk_primitive(PrimitiveType::Long),
2809 ]);
2810 let reader = mk_primitive(PrimitiveType::Bytes);
2811 let mut maker = Maker::new(false, false, Tz::default());
2812 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2813 assert!(matches!(dt.codec(), Codec::Binary));
2814 let resolved = match dt.resolution {
2815 Some(ResolutionInfo::Union(u)) => u,
2816 other => panic!("expected union resolution info, got {other:?}"),
2817 };
2818 assert!(resolved.writer_is_union && !resolved.reader_is_union);
2819 assert_eq!(
2820 resolved.writer_to_reader.as_ref(),
2821 &[
2822 Some((0, ResolutionInfo::Promotion(Promotion::StringToBytes))),
2823 None
2824 ]
2825 );
2826 }
2827
2828 #[test]
2829 fn test_resolve_writer_non_union_to_reader_union_prefers_direct_over_promotion() {
2830 let writer = mk_primitive(PrimitiveType::Long);
2831 let reader = mk_union(vec![
2832 mk_primitive(PrimitiveType::Long),
2833 mk_primitive(PrimitiveType::Double),
2834 ]);
2835 let mut maker = Maker::new(false, false, Tz::default());
2836 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2837 let resolved = match dt.resolution {
2838 Some(ResolutionInfo::Union(u)) => u,
2839 other => panic!("expected union resolution info, got {other:?}"),
2840 };
2841 assert!(!resolved.writer_is_union && resolved.reader_is_union);
2842 assert_eq!(
2843 resolved.writer_to_reader.as_ref(),
2844 &[Some((0, ResolutionInfo::Promotion(Promotion::Direct)))]
2845 );
2846 }
2847
2848 #[test]
2849 fn test_resolve_writer_non_union_to_reader_union_uses_promotion_when_needed() {
2850 let writer = mk_primitive(PrimitiveType::Int);
2851 let reader = mk_union(vec![
2852 mk_primitive(PrimitiveType::Null),
2853 mk_primitive(PrimitiveType::Long),
2854 mk_primitive(PrimitiveType::String),
2855 ]);
2856 let mut maker = Maker::new(false, false, Tz::default());
2857 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2858 let resolved = match dt.resolution {
2859 Some(ResolutionInfo::Union(u)) => u,
2860 other => panic!("expected union resolution info, got {other:?}"),
2861 };
2862 assert_eq!(
2863 resolved.writer_to_reader.as_ref(),
2864 &[Some((1, ResolutionInfo::Promotion(Promotion::IntToLong)))]
2865 );
2866 }
2867
2868 #[test]
2869 fn test_resolve_writer_non_union_to_reader_union_preserves_inner_record_defaults() {
2870 let writer = Schema::Complex(ComplexType::Record(Record {
2874 name: "Inner",
2875 namespace: None,
2876 doc: None,
2877 aliases: vec![],
2878 fields: vec![AvroFieldSchema {
2879 name: "a",
2880 doc: None,
2881 r#type: mk_primitive(PrimitiveType::Int),
2882 default: None,
2883 aliases: vec![],
2884 }],
2885 attributes: Attributes::default(),
2886 }));
2887 let reader = mk_union(vec![
2888 Schema::Complex(ComplexType::Record(Record {
2889 name: "Inner",
2890 namespace: None,
2891 doc: None,
2892 aliases: vec![],
2893 fields: vec![
2894 AvroFieldSchema {
2895 name: "a",
2896 doc: None,
2897 r#type: mk_primitive(PrimitiveType::Int),
2898 default: None,
2899 aliases: vec![],
2900 },
2901 AvroFieldSchema {
2902 name: "b",
2903 doc: None,
2904 r#type: mk_primitive(PrimitiveType::Int),
2905 default: Some(Value::Number(serde_json::Number::from(42))),
2906 aliases: vec![],
2907 },
2908 ],
2909 attributes: Attributes::default(),
2910 })),
2911 mk_primitive(PrimitiveType::String),
2912 ]);
2913 let mut maker = Maker::new(false, false, Default::default());
2914 let dt = maker
2915 .make_data_type(&writer, Some(&reader), None)
2916 .expect("resolution should succeed");
2917 let resolved = match dt.resolution.as_ref() {
2919 Some(ResolutionInfo::Union(u)) => u,
2920 other => panic!("expected union resolution info, got {other:?}"),
2921 };
2922 assert!(!resolved.writer_is_union && resolved.reader_is_union);
2923 assert_eq!(
2924 resolved.writer_to_reader.len(),
2925 1,
2926 "expected the non-union record to resolve to a union variant"
2927 );
2928 let resolution = match resolved.writer_to_reader.first().unwrap() {
2929 Some((0, resolution)) => resolution,
2930 other => panic!("unexpected writer-to-reader table value {other:?}"),
2931 };
2932 match resolution {
2933 ResolutionInfo::Record(ResolvedRecord {
2934 writer_fields,
2935 default_fields,
2936 }) => {
2937 assert_eq!(writer_fields.len(), 1);
2938 assert!(matches!(writer_fields[0], ResolvedField::ToReader(0, _)));
2939 assert_eq!(default_fields.len(), 1);
2940 assert_eq!(default_fields[0], 1);
2941 }
2942 other => panic!("unexpected resolution {other:?}"),
2943 }
2944 let children = match dt.codec() {
2946 Codec::Union(children, _, _) => children,
2947 other => panic!("expected union codec, got {other:?}"),
2948 };
2949 let inner_fields = match children[0].codec() {
2950 Codec::Struct(f) => f,
2951 other => panic!("expected struct codec for Inner, got {other:?}"),
2952 };
2953 assert_eq!(inner_fields.len(), 2);
2954 assert_eq!(inner_fields[1].name(), "b");
2955 assert_eq!(
2956 inner_fields[1].data_type().resolution,
2957 Some(ResolutionInfo::DefaultValue(AvroLiteral::Int(42))),
2958 "field b should have DefaultValue(Int(42)) from schema resolution"
2959 );
2960 }
2961
2962 #[test]
2963 fn test_resolve_writer_union_to_reader_union_preserves_inner_record_defaults() {
2964 let writer = mk_union(vec![
2968 mk_primitive(PrimitiveType::String),
2969 Schema::Complex(ComplexType::Record(Record {
2970 name: "Inner",
2971 namespace: None,
2972 doc: None,
2973 aliases: vec![],
2974 fields: vec![AvroFieldSchema {
2975 name: "a",
2976 doc: None,
2977 r#type: mk_primitive(PrimitiveType::Int),
2978 default: None,
2979 aliases: vec![],
2980 }],
2981 attributes: Attributes::default(),
2982 })),
2983 ]);
2984 let reader = mk_union(vec![
2985 Schema::Complex(ComplexType::Record(Record {
2986 name: "Inner",
2987 namespace: None,
2988 doc: None,
2989 aliases: vec![],
2990 fields: vec![
2991 AvroFieldSchema {
2992 name: "a",
2993 doc: None,
2994 r#type: mk_primitive(PrimitiveType::Int),
2995 default: None,
2996 aliases: vec![],
2997 },
2998 AvroFieldSchema {
2999 name: "b",
3000 doc: None,
3001 r#type: mk_primitive(PrimitiveType::Int),
3002 default: Some(Value::Number(serde_json::Number::from(42))),
3003 aliases: vec![],
3004 },
3005 ],
3006 attributes: Attributes::default(),
3007 })),
3008 mk_primitive(PrimitiveType::String),
3009 ]);
3010 let mut maker = Maker::new(false, false, Default::default());
3011 let dt = maker
3012 .make_data_type(&writer, Some(&reader), None)
3013 .expect("resolution should succeed");
3014 let resolved = match dt.resolution.as_ref() {
3016 Some(ResolutionInfo::Union(u)) => u,
3017 other => panic!("expected union resolution info, got {other:?}"),
3018 };
3019 assert!(resolved.writer_is_union && resolved.reader_is_union);
3020 assert_eq!(resolved.writer_to_reader.len(), 2);
3021 let resolution = match resolved.writer_to_reader[1].as_ref() {
3022 Some((0, resolution)) => resolution,
3023 other => panic!("unexpected writer-to-reader table value {other:?}"),
3024 };
3025 match resolution {
3026 ResolutionInfo::Record(ResolvedRecord {
3027 writer_fields,
3028 default_fields,
3029 }) => {
3030 assert_eq!(writer_fields.len(), 1);
3031 assert!(matches!(writer_fields[0], ResolvedField::ToReader(0, _)));
3032 assert_eq!(default_fields.len(), 1);
3033 assert_eq!(default_fields[0], 1);
3034 }
3035 other => panic!("unexpected resolution {other:?}"),
3036 }
3037 let children = match dt.codec() {
3039 Codec::Union(children, _, _) => children,
3040 other => panic!("expected union codec, got {other:?}"),
3041 };
3042 let inner_fields = match children[0].codec() {
3043 Codec::Struct(f) => f,
3044 other => panic!("expected struct codec for Inner, got {other:?}"),
3045 };
3046 assert_eq!(inner_fields.len(), 2);
3047 assert_eq!(inner_fields[1].name(), "b");
3048 assert_eq!(
3049 inner_fields[1].data_type().resolution,
3050 Some(ResolutionInfo::DefaultValue(AvroLiteral::Int(42))),
3051 "field b should have DefaultValue(Int(42)) from schema resolution"
3052 );
3053 }
3054
3055 #[test]
3056 fn test_resolve_both_nullable_unions_direct_match() {
3057 let writer = mk_union(vec![
3058 mk_primitive(PrimitiveType::Null),
3059 mk_primitive(PrimitiveType::String),
3060 ]);
3061 let reader = mk_union(vec![
3062 mk_primitive(PrimitiveType::String),
3063 mk_primitive(PrimitiveType::Null),
3064 ]);
3065 let mut maker = Maker::new(false, false, Tz::default());
3066 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
3067 assert!(matches!(dt.codec(), Codec::Utf8));
3068 assert_eq!(dt.nullability, Some(Nullability::NullFirst));
3069 assert_eq!(
3070 dt.resolution,
3071 Some(ResolutionInfo::Union(ResolvedUnion {
3072 writer_to_reader: [
3073 None,
3074 Some((0, ResolutionInfo::Promotion(Promotion::Direct)))
3075 ]
3076 .into(),
3077 writer_is_union: true,
3078 reader_is_union: true
3079 }))
3080 );
3081 }
3082
3083 #[test]
3084 fn test_resolve_both_nullable_unions_with_promotion() {
3085 let writer = mk_union(vec![
3086 mk_primitive(PrimitiveType::Null),
3087 mk_primitive(PrimitiveType::Int),
3088 ]);
3089 let reader = mk_union(vec![
3090 mk_primitive(PrimitiveType::Double),
3091 mk_primitive(PrimitiveType::Null),
3092 ]);
3093 let mut maker = Maker::new(false, false, Tz::default());
3094 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
3095 assert!(matches!(dt.codec(), Codec::Float64));
3096 assert_eq!(dt.nullability, Some(Nullability::NullFirst));
3097 assert_eq!(
3098 dt.resolution,
3099 Some(ResolutionInfo::Union(ResolvedUnion {
3100 writer_to_reader: [
3101 None,
3102 Some((0, ResolutionInfo::Promotion(Promotion::IntToDouble)))
3103 ]
3104 .into(),
3105 writer_is_union: true,
3106 reader_is_union: true
3107 }))
3108 );
3109 }
3110
3111 #[test]
3112 fn test_resolve_type_promotion() {
3113 let writer_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3114 let reader_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Long));
3115 let mut maker = Maker::new(false, false, Tz::default());
3116 let result = maker
3117 .make_data_type(&writer_schema, Some(&reader_schema), None)
3118 .unwrap();
3119 assert!(matches!(result.codec, Codec::Int64));
3120 assert_eq!(
3121 result.resolution,
3122 Some(ResolutionInfo::Promotion(Promotion::IntToLong))
3123 );
3124 }
3125
3126 #[test]
3127 fn test_nested_record_type_reuse_without_namespace() {
3128 let schema_str = r#"
3129 {
3130 "type": "record",
3131 "name": "Record",
3132 "fields": [
3133 {
3134 "name": "nested",
3135 "type": {
3136 "type": "record",
3137 "name": "Nested",
3138 "fields": [
3139 { "name": "nested_int", "type": "int" }
3140 ]
3141 }
3142 },
3143 { "name": "nestedRecord", "type": "Nested" },
3144 { "name": "nestedArray", "type": { "type": "array", "items": "Nested" } },
3145 { "name": "nestedMap", "type": { "type": "map", "values": "Nested" } }
3146 ]
3147 }
3148 "#;
3149
3150 let schema: Schema = serde_json::from_str(schema_str).unwrap();
3151
3152 let mut maker = Maker::new(false, false, Tz::default());
3153 let avro_data_type = maker.make_data_type(&schema, None, None).unwrap();
3154
3155 if let Codec::Struct(fields) = avro_data_type.codec() {
3156 assert_eq!(fields.len(), 4);
3157
3158 assert_eq!(fields[0].name(), "nested");
3160 let nested_data_type = fields[0].data_type();
3161 if let Codec::Struct(nested_fields) = nested_data_type.codec() {
3162 assert_eq!(nested_fields.len(), 1);
3163 assert_eq!(nested_fields[0].name(), "nested_int");
3164 assert!(matches!(nested_fields[0].data_type().codec(), Codec::Int32));
3165 } else {
3166 panic!(
3167 "'nested' field is not a struct but {:?}",
3168 nested_data_type.codec()
3169 );
3170 }
3171
3172 assert_eq!(fields[1].name(), "nestedRecord");
3174 let nested_record_data_type = fields[1].data_type();
3175 assert_eq!(
3176 nested_record_data_type.codec().data_type(),
3177 nested_data_type.codec().data_type()
3178 );
3179
3180 assert_eq!(fields[2].name(), "nestedArray");
3182 if let Codec::List(item_type) = fields[2].data_type().codec() {
3183 assert_eq!(
3184 item_type.codec().data_type(),
3185 nested_data_type.codec().data_type()
3186 );
3187 } else {
3188 panic!("'nestedArray' field is not a list");
3189 }
3190
3191 assert_eq!(fields[3].name(), "nestedMap");
3193 if let Codec::Map(value_type) = fields[3].data_type().codec() {
3194 assert_eq!(
3195 value_type.codec().data_type(),
3196 nested_data_type.codec().data_type()
3197 );
3198 } else {
3199 panic!("'nestedMap' field is not a map");
3200 }
3201 } else {
3202 panic!("Top-level schema is not a struct");
3203 }
3204 }
3205
3206 #[test]
3207 fn test_nested_enum_type_reuse_with_namespace() {
3208 let schema_str = r#"
3209 {
3210 "type": "record",
3211 "name": "Record",
3212 "namespace": "record_ns",
3213 "fields": [
3214 {
3215 "name": "status",
3216 "type": {
3217 "type": "enum",
3218 "name": "Status",
3219 "namespace": "enum_ns",
3220 "symbols": ["ACTIVE", "INACTIVE", "PENDING"]
3221 }
3222 },
3223 { "name": "backupStatus", "type": "enum_ns.Status" },
3224 { "name": "statusHistory", "type": { "type": "array", "items": "enum_ns.Status" } },
3225 { "name": "statusMap", "type": { "type": "map", "values": "enum_ns.Status" } }
3226 ]
3227 }
3228 "#;
3229
3230 let schema: Schema = serde_json::from_str(schema_str).unwrap();
3231
3232 let mut maker = Maker::new(false, false, Tz::default());
3233 let avro_data_type = maker.make_data_type(&schema, None, None).unwrap();
3234
3235 if let Codec::Struct(fields) = avro_data_type.codec() {
3236 assert_eq!(fields.len(), 4);
3237
3238 assert_eq!(fields[0].name(), "status");
3240 let status_data_type = fields[0].data_type();
3241 if let Codec::Enum(symbols) = status_data_type.codec() {
3242 assert_eq!(symbols.as_ref(), &["ACTIVE", "INACTIVE", "PENDING"]);
3243 } else {
3244 panic!(
3245 "'status' field is not an enum but {:?}",
3246 status_data_type.codec()
3247 );
3248 }
3249
3250 assert_eq!(fields[1].name(), "backupStatus");
3252 let backup_status_data_type = fields[1].data_type();
3253 assert_eq!(
3254 backup_status_data_type.codec().data_type(),
3255 status_data_type.codec().data_type()
3256 );
3257
3258 assert_eq!(fields[2].name(), "statusHistory");
3260 if let Codec::List(item_type) = fields[2].data_type().codec() {
3261 assert_eq!(
3262 item_type.codec().data_type(),
3263 status_data_type.codec().data_type()
3264 );
3265 } else {
3266 panic!("'statusHistory' field is not a list");
3267 }
3268
3269 assert_eq!(fields[3].name(), "statusMap");
3271 if let Codec::Map(value_type) = fields[3].data_type().codec() {
3272 assert_eq!(
3273 value_type.codec().data_type(),
3274 status_data_type.codec().data_type()
3275 );
3276 } else {
3277 panic!("'statusMap' field is not a map");
3278 }
3279 } else {
3280 panic!("Top-level schema is not a struct");
3281 }
3282 }
3283
3284 #[test]
3285 fn test_resolve_from_writer_and_reader_defaults_root_name_for_non_record_reader() {
3286 let writer_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
3287 let reader_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
3288 let mut maker = Maker::new(false, false, Tz::default());
3289 let data_type = maker
3290 .make_data_type(&writer_schema, Some(&reader_schema), None)
3291 .expect("resolution should succeed");
3292 let field = AvroField {
3293 name: AVRO_ROOT_RECORD_DEFAULT_NAME.to_string(),
3294 data_type,
3295 };
3296 assert_eq!(field.name(), AVRO_ROOT_RECORD_DEFAULT_NAME);
3297 assert!(matches!(field.data_type().codec(), Codec::Utf8));
3298 }
3299
3300 fn json_string(s: &str) -> Value {
3301 Value::String(s.to_string())
3302 }
3303
3304 fn assert_default_stored(dt: &AvroDataType, default_json: &Value) {
3305 let stored = dt
3306 .metadata
3307 .get(AVRO_FIELD_DEFAULT_METADATA_KEY)
3308 .cloned()
3309 .unwrap_or_default();
3310 let expected = serde_json::to_string(default_json).unwrap();
3311 assert_eq!(stored, expected, "stored default metadata should match");
3312 }
3313
3314 #[test]
3315 fn test_validate_and_store_default_null_and_nullability_rules() {
3316 let mut dt_null = AvroDataType::new(Codec::Null, HashMap::new(), None);
3317 let lit = dt_null.parse_and_store_default(&Value::Null).unwrap();
3318 assert_eq!(lit, AvroLiteral::Null);
3319 assert_default_stored(&dt_null, &Value::Null);
3320 let mut dt_int = AvroDataType::new(Codec::Int32, HashMap::new(), None);
3321 let err = dt_int.parse_and_store_default(&Value::Null).unwrap_err();
3322 assert!(
3323 err.to_string()
3324 .contains("JSON null default is only valid for `null` type"),
3325 "unexpected error: {err}"
3326 );
3327 let mut dt_int_nf =
3328 AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullFirst));
3329 let lit2 = dt_int_nf.parse_and_store_default(&Value::Null).unwrap();
3330 assert_eq!(lit2, AvroLiteral::Null);
3331 assert_default_stored(&dt_int_nf, &Value::Null);
3332 let mut dt_int_ns =
3333 AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullSecond));
3334 let err2 = dt_int_ns.parse_and_store_default(&Value::Null).unwrap_err();
3335 assert!(
3336 err2.to_string()
3337 .contains("JSON null default is only valid for `null` type"),
3338 "unexpected error: {err2}"
3339 );
3340 }
3341
3342 #[test]
3343 fn test_validate_and_store_default_primitives_and_temporal() {
3344 let mut dt_bool = AvroDataType::new(Codec::Boolean, HashMap::new(), None);
3345 let lit = dt_bool.parse_and_store_default(&Value::Bool(true)).unwrap();
3346 assert_eq!(lit, AvroLiteral::Boolean(true));
3347 assert_default_stored(&dt_bool, &Value::Bool(true));
3348 let mut dt_i32 = AvroDataType::new(Codec::Int32, HashMap::new(), None);
3349 let lit = dt_i32
3350 .parse_and_store_default(&serde_json::json!(123))
3351 .unwrap();
3352 assert_eq!(lit, AvroLiteral::Int(123));
3353 assert_default_stored(&dt_i32, &serde_json::json!(123));
3354 let err = dt_i32
3355 .parse_and_store_default(&serde_json::json!(i64::from(i32::MAX) + 1))
3356 .unwrap_err();
3357 assert!(format!("{err}").contains("out of i32 range"));
3358 let mut dt_i64 = AvroDataType::new(Codec::Int64, HashMap::new(), None);
3359 let lit = dt_i64
3360 .parse_and_store_default(&serde_json::json!(1234567890))
3361 .unwrap();
3362 assert_eq!(lit, AvroLiteral::Long(1234567890));
3363 assert_default_stored(&dt_i64, &serde_json::json!(1234567890));
3364 let mut dt_f32 = AvroDataType::new(Codec::Float32, HashMap::new(), None);
3365 let lit = dt_f32
3366 .parse_and_store_default(&serde_json::json!(1.25))
3367 .unwrap();
3368 assert_eq!(lit, AvroLiteral::Float(1.25));
3369 assert_default_stored(&dt_f32, &serde_json::json!(1.25));
3370 let err = dt_f32
3371 .parse_and_store_default(&serde_json::json!(1e39))
3372 .unwrap_err();
3373 assert!(format!("{err}").contains("out of f32 range"));
3374 let mut dt_f64 = AvroDataType::new(Codec::Float64, HashMap::new(), None);
3375 let lit = dt_f64
3376 .parse_and_store_default(&serde_json::json!(std::f64::consts::PI))
3377 .unwrap();
3378 assert_eq!(lit, AvroLiteral::Double(std::f64::consts::PI));
3379 assert_default_stored(&dt_f64, &serde_json::json!(std::f64::consts::PI));
3380 let mut dt_str = AvroDataType::new(Codec::Utf8, HashMap::new(), None);
3381 let l = dt_str
3382 .parse_and_store_default(&json_string("hello"))
3383 .unwrap();
3384 assert_eq!(l, AvroLiteral::String("hello".into()));
3385 assert_default_stored(&dt_str, &json_string("hello"));
3386 let mut dt_strv = AvroDataType::new(Codec::Utf8View, HashMap::new(), None);
3387 let l = dt_strv
3388 .parse_and_store_default(&json_string("view"))
3389 .unwrap();
3390 assert_eq!(l, AvroLiteral::String("view".into()));
3391 assert_default_stored(&dt_strv, &json_string("view"));
3392 let mut dt_uuid = AvroDataType::new(Codec::Uuid, HashMap::new(), None);
3393 let l = dt_uuid
3394 .parse_and_store_default(&json_string("00000000-0000-0000-0000-000000000000"))
3395 .unwrap();
3396 assert_eq!(
3397 l,
3398 AvroLiteral::String("00000000-0000-0000-0000-000000000000".into())
3399 );
3400 let mut dt_bin = AvroDataType::new(Codec::Binary, HashMap::new(), None);
3401 let l = dt_bin.parse_and_store_default(&json_string("ABC")).unwrap();
3402 assert_eq!(l, AvroLiteral::Bytes(vec![65, 66, 67]));
3403 let err = dt_bin
3404 .parse_and_store_default(&json_string("€")) .unwrap_err();
3406 assert!(format!("{err}").contains("Invalid codepoint"));
3407 let mut dt_date = AvroDataType::new(Codec::Date32, HashMap::new(), None);
3408 let ld = dt_date
3409 .parse_and_store_default(&serde_json::json!(1))
3410 .unwrap();
3411 assert_eq!(ld, AvroLiteral::Int(1));
3412 let mut dt_tmill = AvroDataType::new(Codec::TimeMillis, HashMap::new(), None);
3413 let lt = dt_tmill
3414 .parse_and_store_default(&serde_json::json!(86_400_000))
3415 .unwrap();
3416 assert_eq!(lt, AvroLiteral::Int(86_400_000));
3417 let mut dt_tmicros = AvroDataType::new(Codec::TimeMicros, HashMap::new(), None);
3418 let ltm = dt_tmicros
3419 .parse_and_store_default(&serde_json::json!(1_000_000))
3420 .unwrap();
3421 assert_eq!(ltm, AvroLiteral::Long(1_000_000));
3422 let mut dt_ts_milli = AvroDataType::new(Codec::TimestampMillis(None), HashMap::new(), None);
3423 let l1 = dt_ts_milli
3424 .parse_and_store_default(&serde_json::json!(123))
3425 .unwrap();
3426 assert_eq!(l1, AvroLiteral::Long(123));
3427 let mut dt_ts_micro = AvroDataType::new(Codec::TimestampMicros(None), HashMap::new(), None);
3428 let l2 = dt_ts_micro
3429 .parse_and_store_default(&serde_json::json!(456))
3430 .unwrap();
3431 assert_eq!(l2, AvroLiteral::Long(456));
3432 }
3433
3434 #[cfg(feature = "avro_custom_types")]
3435 #[test]
3436 fn test_validate_and_store_default_custom_integer_ranges() {
3437 let mut dt_i8 = AvroDataType::new(Codec::Int8, HashMap::new(), None);
3438 let lit_i8 = dt_i8
3439 .parse_and_store_default(&serde_json::json!(i8::MAX))
3440 .unwrap();
3441 assert_eq!(lit_i8, AvroLiteral::Int(i8::MAX as i32));
3442 let err_i8_high = dt_i8
3443 .parse_and_store_default(&serde_json::json!(i8::MAX as i64 + 1))
3444 .unwrap_err();
3445 assert!(err_i8_high.to_string().contains("out of i8 range"));
3446 let err_i8_low = dt_i8
3447 .parse_and_store_default(&serde_json::json!(i8::MIN as i64 - 1))
3448 .unwrap_err();
3449 assert!(err_i8_low.to_string().contains("out of i8 range"));
3450
3451 let mut dt_i16 = AvroDataType::new(Codec::Int16, HashMap::new(), None);
3452 let lit_i16 = dt_i16
3453 .parse_and_store_default(&serde_json::json!(i16::MIN))
3454 .unwrap();
3455 assert_eq!(lit_i16, AvroLiteral::Int(i16::MIN as i32));
3456 let err_i16_high = dt_i16
3457 .parse_and_store_default(&serde_json::json!(i16::MAX as i64 + 1))
3458 .unwrap_err();
3459 assert!(err_i16_high.to_string().contains("out of i16 range"));
3460 let err_i16_low = dt_i16
3461 .parse_and_store_default(&serde_json::json!(i16::MIN as i64 - 1))
3462 .unwrap_err();
3463 assert!(err_i16_low.to_string().contains("out of i16 range"));
3464
3465 let mut dt_u8 = AvroDataType::new(Codec::UInt8, HashMap::new(), None);
3466 let lit_u8 = dt_u8
3467 .parse_and_store_default(&serde_json::json!(u8::MAX))
3468 .unwrap();
3469 assert_eq!(lit_u8, AvroLiteral::Int(u8::MAX as i32));
3470 let err_u8_neg = dt_u8
3471 .parse_and_store_default(&serde_json::json!(-1))
3472 .unwrap_err();
3473 assert!(err_u8_neg.to_string().contains("out of u8 range"));
3474 let err_u8_high = dt_u8
3475 .parse_and_store_default(&serde_json::json!(u8::MAX as i64 + 1))
3476 .unwrap_err();
3477 assert!(err_u8_high.to_string().contains("out of u8 range"));
3478
3479 let mut dt_u16 = AvroDataType::new(Codec::UInt16, HashMap::new(), None);
3480 let lit_u16 = dt_u16
3481 .parse_and_store_default(&serde_json::json!(u16::MAX))
3482 .unwrap();
3483 assert_eq!(lit_u16, AvroLiteral::Int(u16::MAX as i32));
3484 let err_u16_neg = dt_u16
3485 .parse_and_store_default(&serde_json::json!(-1))
3486 .unwrap_err();
3487 assert!(err_u16_neg.to_string().contains("out of u16 range"));
3488 let err_u16_high = dt_u16
3489 .parse_and_store_default(&serde_json::json!(u16::MAX as i64 + 1))
3490 .unwrap_err();
3491 assert!(err_u16_high.to_string().contains("out of u16 range"));
3492
3493 let mut dt_u32 = AvroDataType::new(Codec::UInt32, HashMap::new(), None);
3494 let lit_u32 = dt_u32
3495 .parse_and_store_default(&serde_json::json!(u32::MAX as i64))
3496 .unwrap();
3497 assert_eq!(lit_u32, AvroLiteral::Long(u32::MAX as i64));
3498 let err_u32_neg = dt_u32
3499 .parse_and_store_default(&serde_json::json!(-1))
3500 .unwrap_err();
3501 assert!(err_u32_neg.to_string().contains("out of u32 range"));
3502 let err_u32_high = dt_u32
3503 .parse_and_store_default(&serde_json::json!(u32::MAX as i64 + 1))
3504 .unwrap_err();
3505 assert!(err_u32_high.to_string().contains("out of u32 range"));
3506 }
3507
3508 #[test]
3509 fn test_validate_and_store_default_fixed_decimal_interval() {
3510 let mut dt_fixed = AvroDataType::new(Codec::Fixed(4), HashMap::new(), None);
3511 let l = dt_fixed
3512 .parse_and_store_default(&json_string("WXYZ"))
3513 .unwrap();
3514 assert_eq!(l, AvroLiteral::Bytes(vec![87, 88, 89, 90]));
3515 let err = dt_fixed
3516 .parse_and_store_default(&json_string("TOO LONG"))
3517 .unwrap_err();
3518 assert!(err.to_string().contains("Default length"));
3519 let mut dt_dec_fixed =
3520 AvroDataType::new(Codec::Decimal(10, Some(2), Some(3)), HashMap::new(), None);
3521 let l = dt_dec_fixed
3522 .parse_and_store_default(&json_string("abc"))
3523 .unwrap();
3524 assert_eq!(l, AvroLiteral::Bytes(vec![97, 98, 99]));
3525 let err = dt_dec_fixed
3526 .parse_and_store_default(&json_string("toolong"))
3527 .unwrap_err();
3528 assert!(err.to_string().contains("Default length"));
3529 let mut dt_dec_bytes =
3530 AvroDataType::new(Codec::Decimal(10, Some(2), None), HashMap::new(), None);
3531 let l = dt_dec_bytes
3532 .parse_and_store_default(&json_string("freeform"))
3533 .unwrap();
3534 assert_eq!(
3535 l,
3536 AvroLiteral::Bytes("freeform".bytes().collect::<Vec<_>>())
3537 );
3538 let mut dt_interval = AvroDataType::new(Codec::Interval, HashMap::new(), None);
3539 let l = dt_interval
3540 .parse_and_store_default(&json_string("ABCDEFGHIJKL"))
3541 .unwrap();
3542 assert_eq!(
3543 l,
3544 AvroLiteral::Bytes("ABCDEFGHIJKL".bytes().collect::<Vec<_>>())
3545 );
3546 let err = dt_interval
3547 .parse_and_store_default(&json_string("short"))
3548 .unwrap_err();
3549 assert!(err.to_string().contains("Default length"));
3550 }
3551
3552 #[test]
3553 fn test_validate_and_store_default_enum_list_map_struct() {
3554 let symbols: Arc<[String]> = ["RED".to_string(), "GREEN".to_string(), "BLUE".to_string()]
3555 .into_iter()
3556 .collect();
3557 let mut dt_enum = AvroDataType::new(Codec::Enum(symbols), HashMap::new(), None);
3558 let l = dt_enum
3559 .parse_and_store_default(&json_string("GREEN"))
3560 .unwrap();
3561 assert_eq!(l, AvroLiteral::Enum("GREEN".into()));
3562 let err = dt_enum
3563 .parse_and_store_default(&json_string("YELLOW"))
3564 .unwrap_err();
3565 assert!(err.to_string().contains("Default enum symbol"));
3566 let item = AvroDataType::new(Codec::Int64, HashMap::new(), None);
3567 let mut dt_list = AvroDataType::new(Codec::List(Arc::new(item)), HashMap::new(), None);
3568 let val = serde_json::json!([1, 2, 3]);
3569 let l = dt_list.parse_and_store_default(&val).unwrap();
3570 assert_eq!(
3571 l,
3572 AvroLiteral::Array(vec![
3573 AvroLiteral::Long(1),
3574 AvroLiteral::Long(2),
3575 AvroLiteral::Long(3)
3576 ])
3577 );
3578 let err = dt_list
3579 .parse_and_store_default(&serde_json::json!({"not":"array"}))
3580 .unwrap_err();
3581 assert!(err.to_string().contains("JSON array"));
3582 let val_dt = AvroDataType::new(Codec::Float64, HashMap::new(), None);
3583 let mut dt_map = AvroDataType::new(Codec::Map(Arc::new(val_dt)), HashMap::new(), None);
3584 let mv = serde_json::json!({"x": 1.5, "y": 2.5});
3585 let l = dt_map.parse_and_store_default(&mv).unwrap();
3586 let mut expected = IndexMap::new();
3587 expected.insert("x".into(), AvroLiteral::Double(1.5));
3588 expected.insert("y".into(), AvroLiteral::Double(2.5));
3589 assert_eq!(l, AvroLiteral::Map(expected));
3590 let err = dt_map
3592 .parse_and_store_default(&serde_json::json!(123))
3593 .unwrap_err();
3594 assert!(err.to_string().contains("JSON object"));
3595 let mut field_a = AvroField {
3596 name: "a".into(),
3597 data_type: AvroDataType::new(Codec::Int32, HashMap::new(), None),
3598 };
3599 let field_b = AvroField {
3600 name: "b".into(),
3601 data_type: AvroDataType::new(
3602 Codec::Int64,
3603 HashMap::new(),
3604 Some(Nullability::NullFirst),
3605 ),
3606 };
3607 let mut c_md = HashMap::new();
3608 c_md.insert(AVRO_FIELD_DEFAULT_METADATA_KEY.into(), "\"xyz\"".into());
3609 let field_c = AvroField {
3610 name: "c".into(),
3611 data_type: AvroDataType::new(Codec::Utf8, c_md, None),
3612 };
3613 field_a.data_type.metadata.insert("doc".into(), "na".into());
3614 let struct_fields: Arc<[AvroField]> = Arc::from(vec![field_a, field_b, field_c]);
3615 let mut dt_struct = AvroDataType::new(Codec::Struct(struct_fields), HashMap::new(), None);
3616 let default_obj = serde_json::json!({"a": 7});
3617 let l = dt_struct.parse_and_store_default(&default_obj).unwrap();
3618 let mut expected = IndexMap::new();
3619 expected.insert("a".into(), AvroLiteral::Int(7));
3620 expected.insert("b".into(), AvroLiteral::Null);
3621 expected.insert("c".into(), AvroLiteral::String("xyz".into()));
3622 assert_eq!(l, AvroLiteral::Map(expected));
3623 assert_default_stored(&dt_struct, &default_obj);
3624 let req_field = AvroField {
3625 name: "req".into(),
3626 data_type: AvroDataType::new(Codec::Boolean, HashMap::new(), None),
3627 };
3628 let mut dt_bad = AvroDataType::new(
3629 Codec::Struct(Arc::from(vec![req_field])),
3630 HashMap::new(),
3631 None,
3632 );
3633 let err = dt_bad
3634 .parse_and_store_default(&serde_json::json!({}))
3635 .unwrap_err();
3636 assert!(
3637 err.to_string().contains("missing required subfield 'req'"),
3638 "unexpected error: {err}"
3639 );
3640 let err = dt_struct
3641 .parse_and_store_default(&serde_json::json!(10))
3642 .unwrap_err();
3643 err.to_string().contains("must be a JSON object");
3644 }
3645
3646 #[test]
3647 fn test_resolve_array_promotion_and_reader_metadata() {
3648 let mut w_add: HashMap<&str, Value> = HashMap::new();
3649 w_add.insert("who", json_string("writer"));
3650 let mut r_add: HashMap<&str, Value> = HashMap::new();
3651 r_add.insert("who", json_string("reader"));
3652 let writer_schema = Schema::Complex(ComplexType::Array(Array {
3653 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
3654 attributes: Attributes {
3655 logical_type: None,
3656 additional: w_add,
3657 },
3658 }));
3659 let reader_schema = Schema::Complex(ComplexType::Array(Array {
3660 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Long))),
3661 attributes: Attributes {
3662 logical_type: None,
3663 additional: r_add,
3664 },
3665 }));
3666 let mut maker = Maker::new(false, false, Tz::default());
3667 let dt = maker
3668 .make_data_type(&writer_schema, Some(&reader_schema), None)
3669 .unwrap();
3670 assert_eq!(dt.metadata.get("who"), Some(&"\"reader\"".to_string()));
3671 if let Codec::List(inner) = dt.codec() {
3672 assert!(matches!(inner.codec(), Codec::Int64));
3673 assert_eq!(
3674 inner.resolution,
3675 Some(ResolutionInfo::Promotion(Promotion::IntToLong))
3676 );
3677 } else {
3678 panic!("expected list codec");
3679 }
3680 }
3681
3682 #[test]
3683 fn test_resolve_array_writer_nonunion_items_reader_nullable_items() {
3684 let writer_schema = Schema::Complex(ComplexType::Array(Array {
3685 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
3686 attributes: Attributes::default(),
3687 }));
3688 let reader_schema = Schema::Complex(ComplexType::Array(Array {
3689 items: Box::new(mk_union(vec![
3690 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
3691 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3692 ])),
3693 attributes: Attributes::default(),
3694 }));
3695 let mut maker = Maker::new(false, false, Tz::default());
3696 let dt = maker
3697 .make_data_type(&writer_schema, Some(&reader_schema), None)
3698 .unwrap();
3699 if let Codec::List(inner) = dt.codec() {
3700 assert_eq!(inner.nullability(), Some(Nullability::NullFirst));
3701 assert!(matches!(inner.codec(), Codec::Int32));
3702 match inner.resolution.as_ref() {
3703 Some(ResolutionInfo::Promotion(Promotion::Direct)) => {}
3704 other => panic!("expected Union resolution, got {other:?}"),
3705 }
3706 } else {
3707 panic!("expected List codec");
3708 }
3709 }
3710
3711 #[test]
3712 fn test_resolve_fixed_success_name_and_size_match_and_alias() {
3713 let writer_schema = Schema::Complex(ComplexType::Fixed(Fixed {
3714 name: "MD5",
3715 namespace: None,
3716 aliases: vec!["Hash16"],
3717 size: 16,
3718 attributes: Attributes::default(),
3719 }));
3720 let reader_schema = Schema::Complex(ComplexType::Fixed(Fixed {
3721 name: "Hash16",
3722 namespace: None,
3723 aliases: vec![],
3724 size: 16,
3725 attributes: Attributes::default(),
3726 }));
3727 let mut maker = Maker::new(false, false, Tz::default());
3728 let dt = maker
3729 .make_data_type(&writer_schema, Some(&reader_schema), None)
3730 .unwrap();
3731 assert!(matches!(dt.codec(), Codec::Fixed(16)));
3732 }
3733
3734 #[cfg(feature = "avro_custom_types")]
3735 #[test]
3736 fn test_interval_month_day_nano_custom_logical_type_fixed16() {
3737 let schema = Schema::Complex(ComplexType::Fixed(Fixed {
3738 name: "ArrowIntervalMDN",
3739 namespace: None,
3740 aliases: vec![],
3741 size: 16,
3742 attributes: Attributes {
3743 logical_type: Some("arrow.interval-month-day-nano"),
3744 additional: Default::default(),
3745 },
3746 }));
3747 let mut maker = Maker::new(false, false, Default::default());
3748 let dt = maker.make_data_type(&schema, None, None).unwrap();
3749 assert!(matches!(dt.codec(), Codec::IntervalMonthDayNano));
3750 assert_eq!(
3751 dt.codec.data_type(),
3752 DataType::Interval(IntervalUnit::MonthDayNano)
3753 );
3754 }
3755
3756 #[test]
3757 fn test_resolve_records_mapping_default_fields_and_skip_fields() {
3758 let writer = Schema::Complex(ComplexType::Record(Record {
3759 name: "R",
3760 namespace: None,
3761 doc: None,
3762 aliases: vec![],
3763 fields: vec![
3764 crate::schema::Field {
3765 name: "a",
3766 doc: None,
3767 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3768 default: None,
3769 aliases: vec![],
3770 },
3771 crate::schema::Field {
3772 name: "skipme",
3773 doc: None,
3774 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3775 default: None,
3776 aliases: vec![],
3777 },
3778 crate::schema::Field {
3779 name: "b",
3780 doc: None,
3781 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3782 default: None,
3783 aliases: vec![],
3784 },
3785 ],
3786 attributes: Attributes::default(),
3787 }));
3788 let reader = Schema::Complex(ComplexType::Record(Record {
3789 name: "R",
3790 namespace: None,
3791 doc: None,
3792 aliases: vec![],
3793 fields: vec![
3794 crate::schema::Field {
3795 name: "b",
3796 doc: None,
3797 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3798 default: None,
3799 aliases: vec![],
3800 },
3801 crate::schema::Field {
3802 name: "a",
3803 doc: None,
3804 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3805 default: None,
3806 aliases: vec![],
3807 },
3808 crate::schema::Field {
3809 name: "name",
3810 doc: None,
3811 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3812 default: Some(json_string("anon")),
3813 aliases: vec![],
3814 },
3815 crate::schema::Field {
3816 name: "opt",
3817 doc: None,
3818 r#type: Schema::Union(vec![
3819 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
3820 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3821 ]),
3822 default: None, aliases: vec![],
3824 },
3825 ],
3826 attributes: Attributes::default(),
3827 }));
3828 let mut maker = Maker::new(false, false, Tz::default());
3829 let dt = maker
3830 .make_data_type(&writer, Some(&reader), None)
3831 .expect("record resolution");
3832 let fields = match dt.codec() {
3833 Codec::Struct(f) => f,
3834 other => panic!("expected struct, got {other:?}"),
3835 };
3836 assert_eq!(fields.len(), 4);
3837 assert_eq!(fields[0].name(), "b");
3838 assert_eq!(fields[1].name(), "a");
3839 assert_eq!(fields[2].name(), "name");
3840 assert_eq!(fields[3].name(), "opt");
3841 assert!(matches!(
3842 fields[1].data_type().resolution,
3843 Some(ResolutionInfo::Promotion(Promotion::IntToLong))
3844 ));
3845 let rec = match dt.resolution {
3846 Some(ResolutionInfo::Record(ref r)) => r.clone(),
3847 other => panic!("expected record resolution, got {other:?}"),
3848 };
3849 assert!(matches!(
3850 &rec.writer_fields[..],
3851 &[
3852 ResolvedField::ToReader(1, _),
3853 ResolvedField::Skip(_),
3854 ResolvedField::ToReader(0, _),
3855 ]
3856 ));
3857 assert_eq!(rec.default_fields.as_ref(), &[2usize, 3usize]);
3858 let ResolvedField::Skip(skip1) = &rec.writer_fields[1] else {
3859 panic!("should skip field 1")
3860 };
3861 assert!(matches!(skip1.codec(), Codec::Utf8));
3862 let name_md = &fields[2].data_type().metadata;
3863 assert_eq!(
3864 name_md.get(AVRO_FIELD_DEFAULT_METADATA_KEY),
3865 Some(&"\"anon\"".to_string())
3866 );
3867 let opt_md = &fields[3].data_type().metadata;
3868 assert_eq!(
3869 opt_md.get(AVRO_FIELD_DEFAULT_METADATA_KEY),
3870 Some(&"null".to_string())
3871 );
3872 }
3873
3874 #[test]
3875 fn test_named_type_alias_resolution_record_cross_namespace() {
3876 let writer_record = Record {
3877 name: "PersonV2",
3878 namespace: Some("com.example.v2"),
3879 doc: None,
3880 aliases: vec!["com.example.Person"],
3881 fields: vec![
3882 AvroFieldSchema {
3883 name: "name",
3884 doc: None,
3885 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3886 default: None,
3887 aliases: vec![],
3888 },
3889 AvroFieldSchema {
3890 name: "age",
3891 doc: None,
3892 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3893 default: None,
3894 aliases: vec![],
3895 },
3896 ],
3897 attributes: Attributes::default(),
3898 };
3899 let reader_record = Record {
3900 name: "Person",
3901 namespace: Some("com.example"),
3902 doc: None,
3903 aliases: vec![],
3904 fields: writer_record.fields.clone(),
3905 attributes: Attributes::default(),
3906 };
3907 let writer_schema = Schema::Complex(ComplexType::Record(writer_record));
3908 let reader_schema = Schema::Complex(ComplexType::Record(reader_record));
3909 let mut maker = Maker::new(false, false, Tz::default());
3910 let result = maker
3911 .make_data_type(&writer_schema, Some(&reader_schema), None)
3912 .expect("record alias resolution should succeed");
3913 match result.codec {
3914 Codec::Struct(ref fields) => assert_eq!(fields.len(), 2),
3915 other => panic!("expected struct, got {other:?}"),
3916 }
3917 }
3918
3919 #[test]
3920 fn test_named_type_alias_resolution_enum_cross_namespace() {
3921 let writer_enum = Enum {
3922 name: "ColorV2",
3923 namespace: Some("org.example.v2"),
3924 doc: None,
3925 aliases: vec!["org.example.Color"],
3926 symbols: vec!["RED", "GREEN", "BLUE"],
3927 default: None,
3928 attributes: Attributes::default(),
3929 };
3930 let reader_enum = Enum {
3931 name: "Color",
3932 namespace: Some("org.example"),
3933 doc: None,
3934 aliases: vec![],
3935 symbols: vec!["RED", "GREEN", "BLUE"],
3936 default: None,
3937 attributes: Attributes::default(),
3938 };
3939 let writer_schema = Schema::Complex(ComplexType::Enum(writer_enum));
3940 let reader_schema = Schema::Complex(ComplexType::Enum(reader_enum));
3941 let mut maker = Maker::new(false, false, Tz::default());
3942 maker
3943 .make_data_type(&writer_schema, Some(&reader_schema), None)
3944 .expect("enum alias resolution should succeed");
3945 }
3946
3947 #[test]
3948 fn test_named_type_alias_resolution_fixed_cross_namespace() {
3949 let writer_fixed = Fixed {
3950 name: "Fx10V2",
3951 namespace: Some("ns.v2"),
3952 aliases: vec!["ns.Fx10"],
3953 size: 10,
3954 attributes: Attributes::default(),
3955 };
3956 let reader_fixed = Fixed {
3957 name: "Fx10",
3958 namespace: Some("ns"),
3959 aliases: vec![],
3960 size: 10,
3961 attributes: Attributes::default(),
3962 };
3963 let writer_schema = Schema::Complex(ComplexType::Fixed(writer_fixed));
3964 let reader_schema = Schema::Complex(ComplexType::Fixed(reader_fixed));
3965 let mut maker = Maker::new(false, false, Tz::default());
3966 maker
3967 .make_data_type(&writer_schema, Some(&reader_schema), None)
3968 .expect("fixed alias resolution should succeed");
3969 }
3970}