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