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("run_ends", run_ends_dt, false)),
964 Arc::new(Field::new("values", values.codec().data_type(), true)),
965 )
966 }
967 #[cfg(feature = "avro_custom_types")]
968 Self::Int8 => DataType::Int8,
969 #[cfg(feature = "avro_custom_types")]
970 Self::Int16 => DataType::Int16,
971 #[cfg(feature = "avro_custom_types")]
972 Self::UInt8 => DataType::UInt8,
973 #[cfg(feature = "avro_custom_types")]
974 Self::UInt16 => DataType::UInt16,
975 #[cfg(feature = "avro_custom_types")]
976 Self::UInt32 => DataType::UInt32,
977 #[cfg(feature = "avro_custom_types")]
978 Self::UInt64 => DataType::UInt64,
979 #[cfg(feature = "avro_custom_types")]
980 Self::Float16 => DataType::Float16,
981 #[cfg(feature = "avro_custom_types")]
982 Self::Date64 => DataType::Date64,
983 #[cfg(feature = "avro_custom_types")]
984 Self::TimeNanos => DataType::Time64(TimeUnit::Nanosecond),
985 #[cfg(feature = "avro_custom_types")]
986 Self::Time32Secs => DataType::Time32(TimeUnit::Second),
987 #[cfg(feature = "avro_custom_types")]
988 Self::TimestampSecs(is_utc) => {
989 DataType::Timestamp(TimeUnit::Second, is_utc.then(|| "+00:00".into()))
990 }
991 #[cfg(feature = "avro_custom_types")]
992 Self::IntervalYearMonth => DataType::Interval(IntervalUnit::YearMonth),
993 #[cfg(feature = "avro_custom_types")]
994 Self::IntervalMonthDayNano => DataType::Interval(IntervalUnit::MonthDayNano),
995 #[cfg(feature = "avro_custom_types")]
996 Self::IntervalDayTime => DataType::Interval(IntervalUnit::DayTime),
997 }
998 }
999
1000 pub(crate) fn with_utf8view(self, use_utf8view: bool) -> Self {
1006 if use_utf8view && matches!(self, Self::Utf8) {
1007 Self::Utf8View
1008 } else {
1009 self
1010 }
1011 }
1012
1013 #[inline]
1014 fn union_field_name(&self) -> String {
1015 UnionFieldKind::from(self).as_ref().to_owned()
1016 }
1017}
1018
1019impl From<PrimitiveType> for Codec {
1020 fn from(value: PrimitiveType) -> Self {
1021 match value {
1022 PrimitiveType::Null => Self::Null,
1023 PrimitiveType::Boolean => Self::Boolean,
1024 PrimitiveType::Int => Self::Int32,
1025 PrimitiveType::Long => Self::Int64,
1026 PrimitiveType::Float => Self::Float32,
1027 PrimitiveType::Double => Self::Float64,
1028 PrimitiveType::Bytes => Self::Binary,
1029 PrimitiveType::String => Self::Utf8,
1030 }
1031 }
1032}
1033
1034const fn max_precision_for_fixed_bytes(n: usize) -> Option<usize> {
1043 const MAX_P: [usize; 32] = [
1048 2, 4, 6, 9, 11, 14, 16, 18, 21, 23, 26, 28, 31, 33, 35, 38, 40, 43, 45, 47, 50, 52, 55, 57,
1049 59, 62, 64, 67, 69, 71, 74, 76,
1050 ];
1051 match n {
1052 1..=32 => Some(MAX_P[n - 1]),
1053 _ => None,
1054 }
1055}
1056
1057fn parse_decimal_attributes(
1058 attributes: &Attributes,
1059 fallback_size: Option<usize>,
1060 precision_required: bool,
1061) -> Result<(usize, usize, Option<usize>), ArrowError> {
1062 let precision = attributes
1063 .additional
1064 .get("precision")
1065 .and_then(|v| v.as_u64())
1066 .or(if precision_required { None } else { Some(10) })
1067 .ok_or_else(|| ArrowError::ParseError("Decimal requires precision".to_string()))?
1068 as usize;
1069 let scale = attributes
1070 .additional
1071 .get("scale")
1072 .and_then(|v| v.as_u64())
1073 .unwrap_or(0) as usize;
1074 let size = attributes
1075 .additional
1076 .get("size")
1077 .and_then(|v| v.as_u64())
1078 .map(|s| s as usize)
1079 .or(fallback_size);
1080 if precision == 0 {
1081 return Err(ArrowError::ParseError(
1082 "Decimal requires precision > 0".to_string(),
1083 ));
1084 }
1085 if scale > precision {
1086 return Err(ArrowError::ParseError(format!(
1087 "Decimal has invalid scale > precision: scale={scale}, precision={precision}"
1088 )));
1089 }
1090 if precision > DECIMAL256_MAX_PRECISION as usize {
1091 return Err(ArrowError::ParseError(format!(
1092 "Decimal precision {precision} exceeds maximum supported by Arrow ({DECIMAL256_MAX_PRECISION})"
1093 )));
1094 }
1095 if let Some(sz) = size {
1096 let max_p = max_precision_for_fixed_bytes(sz).ok_or_else(|| {
1097 ArrowError::ParseError(format!(
1098 "Invalid fixed size for decimal: {sz}, must be between 1 and 32 bytes"
1099 ))
1100 })?;
1101 if precision > max_p {
1102 return Err(ArrowError::ParseError(format!(
1103 "Decimal precision {precision} exceeds capacity of fixed size {sz} bytes (max {max_p})"
1104 )));
1105 }
1106 }
1107 Ok((precision, scale, size))
1108}
1109
1110#[derive(Debug, Clone, Copy, PartialEq, Eq, AsRefStr)]
1111#[strum(serialize_all = "snake_case")]
1112enum UnionFieldKind {
1113 Null,
1114 Boolean,
1115 Int,
1116 Long,
1117 Float,
1118 Double,
1119 Bytes,
1120 String,
1121 Date,
1122 TimeMillis,
1123 TimeMicros,
1124 TimestampMillisUtc,
1125 TimestampMillisLocal,
1126 TimestampMicrosUtc,
1127 TimestampMicrosLocal,
1128 TimestampNanosUtc,
1129 TimestampNanosLocal,
1130 Duration,
1131 Fixed,
1132 Decimal,
1133 Enum,
1134 Array,
1135 Record,
1136 Map,
1137 Uuid,
1138 Union,
1139}
1140
1141impl From<&Codec> for UnionFieldKind {
1142 fn from(c: &Codec) -> Self {
1143 match c {
1144 Codec::Null => Self::Null,
1145 Codec::Boolean => Self::Boolean,
1146 Codec::Int32 => Self::Int,
1147 Codec::Int64 => Self::Long,
1148 Codec::Float32 => Self::Float,
1149 Codec::Float64 => Self::Double,
1150 Codec::Binary => Self::Bytes,
1151 Codec::Utf8 | Codec::Utf8View => Self::String,
1152 Codec::Date32 => Self::Date,
1153 Codec::TimeMillis => Self::TimeMillis,
1154 Codec::TimeMicros => Self::TimeMicros,
1155 Codec::TimestampMillis(Some(Tz::OffsetZero)) => Self::TimestampMillisUtc,
1156 Codec::TimestampMillis(Some(Tz::Utc)) => Self::TimestampMillisUtc,
1157 Codec::TimestampMillis(None) => Self::TimestampMillisLocal,
1158 Codec::TimestampMicros(Some(Tz::OffsetZero)) => Self::TimestampMicrosUtc,
1159 Codec::TimestampMicros(Some(Tz::Utc)) => Self::TimestampMicrosUtc,
1160 Codec::TimestampMicros(None) => Self::TimestampMicrosLocal,
1161 Codec::TimestampNanos(Some(Tz::OffsetZero)) => Self::TimestampNanosUtc,
1162 Codec::TimestampNanos(Some(Tz::Utc)) => Self::TimestampNanosUtc,
1163 Codec::TimestampNanos(None) => Self::TimestampNanosLocal,
1164 Codec::Interval => Self::Duration,
1165 Codec::Fixed(_) => Self::Fixed,
1166 Codec::Decimal(..) => Self::Decimal,
1167 Codec::Enum(_) => Self::Enum,
1168 Codec::List(_) => Self::Array,
1169 Codec::Struct(_) => Self::Record,
1170 Codec::Map(_) => Self::Map,
1171 Codec::Uuid => Self::Uuid,
1172 Codec::Union(..) => Self::Union,
1173 #[cfg(feature = "avro_custom_types")]
1174 Codec::RunEndEncoded(values, _) => UnionFieldKind::from(values.codec()),
1175 #[cfg(feature = "avro_custom_types")]
1176 Codec::DurationNanos
1177 | Codec::DurationMicros
1178 | Codec::DurationMillis
1179 | Codec::DurationSeconds => Self::Duration,
1180 #[cfg(feature = "avro_custom_types")]
1181 Codec::Int8 | Codec::Int16 | Codec::UInt8 | Codec::UInt16 => Self::Int,
1182 #[cfg(feature = "avro_custom_types")]
1183 Codec::UInt32 | Codec::Date64 | Codec::TimeNanos | Codec::TimestampSecs(_) => {
1184 Self::Long
1185 }
1186 #[cfg(feature = "avro_custom_types")]
1187 Codec::Time32Secs => Self::TimeMillis, #[cfg(feature = "avro_custom_types")]
1189 Codec::UInt64
1190 | Codec::Float16
1191 | Codec::IntervalYearMonth
1192 | Codec::IntervalMonthDayNano
1193 | Codec::IntervalDayTime => Self::Fixed,
1194 }
1195 }
1196}
1197
1198fn union_branch_name(dt: &AvroDataType) -> String {
1199 if let Some(name) = dt.metadata.get(AVRO_NAME_METADATA_KEY) {
1200 if name.contains('.') {
1201 return name.clone();
1203 }
1204 if let Some(ns) = dt.metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1205 return format!("{ns}.{name}");
1206 }
1207 return name.clone();
1208 }
1209 dt.codec.union_field_name()
1210}
1211
1212fn build_union_fields(encodings: &[AvroDataType]) -> Result<UnionFields, ArrowError> {
1213 let arrow_fields: Vec<Field> = encodings
1214 .iter()
1215 .map(|encoding| encoding.field_with_name(&union_branch_name(encoding)))
1216 .collect();
1217 let type_ids: Vec<i8> = (0..arrow_fields.len()).map(|i| i as i8).collect();
1218 UnionFields::try_new(type_ids, arrow_fields)
1219}
1220
1221#[derive(Debug, Default)]
1225struct Resolver<'a> {
1226 map: HashMap<(&'a str, &'a str), AvroDataType>,
1227}
1228
1229impl<'a> Resolver<'a> {
1230 fn register(&mut self, name: &'a str, namespace: Option<&'a str>, schema: AvroDataType) {
1231 self.map.insert((namespace.unwrap_or(""), name), schema);
1232 }
1233
1234 fn resolve(&self, name: &str, namespace: Option<&'a str>) -> Result<AvroDataType, ArrowError> {
1235 let (namespace, name) = name
1236 .rsplit_once('.')
1237 .unwrap_or_else(|| (namespace.unwrap_or(""), name));
1238 self.map
1239 .get(&(namespace, name))
1240 .ok_or_else(|| ArrowError::ParseError(format!("Failed to resolve {namespace}.{name}")))
1241 .cloned()
1242 }
1243}
1244
1245fn full_name_set(name: &str, ns: Option<&str>, aliases: &[&str]) -> HashSet<String> {
1246 let mut out = HashSet::with_capacity(1 + aliases.len());
1247 let (full, _) = make_full_name(name, ns, None);
1248 out.insert(full);
1249 for a in aliases {
1250 let (fa, _) = make_full_name(a, None, ns);
1251 out.insert(fa);
1252 }
1253 out
1254}
1255
1256fn names_match(
1257 writer_name: &str,
1258 writer_namespace: Option<&str>,
1259 writer_aliases: &[&str],
1260 reader_name: &str,
1261 reader_namespace: Option<&str>,
1262 reader_aliases: &[&str],
1263) -> bool {
1264 let writer_set = full_name_set(writer_name, writer_namespace, writer_aliases);
1265 let reader_set = full_name_set(reader_name, reader_namespace, reader_aliases);
1266 !writer_set.is_disjoint(&reader_set)
1268}
1269
1270fn ensure_names_match(
1271 data_type: &str,
1272 writer_name: &str,
1273 writer_namespace: Option<&str>,
1274 writer_aliases: &[&str],
1275 reader_name: &str,
1276 reader_namespace: Option<&str>,
1277 reader_aliases: &[&str],
1278) -> Result<(), ArrowError> {
1279 if names_match(
1280 writer_name,
1281 writer_namespace,
1282 writer_aliases,
1283 reader_name,
1284 reader_namespace,
1285 reader_aliases,
1286 ) {
1287 Ok(())
1288 } else {
1289 Err(ArrowError::ParseError(format!(
1290 "{data_type} name mismatch writer={writer_name}, reader={reader_name}"
1291 )))
1292 }
1293}
1294
1295fn primitive_of(schema: &Schema) -> Option<PrimitiveType> {
1296 match schema {
1297 Schema::TypeName(TypeName::Primitive(primitive)) => Some(*primitive),
1298 Schema::Type(Type {
1299 r#type: TypeName::Primitive(primitive),
1300 ..
1301 }) => Some(*primitive),
1302 _ => None,
1303 }
1304}
1305
1306fn nullable_union_variants<'x, 'y>(
1307 variant: &'y [Schema<'x>],
1308) -> Option<(Nullability, &'y Schema<'x>)> {
1309 if variant.len() != 2 {
1310 return None;
1311 }
1312 let is_null = |schema: &Schema<'x>| {
1313 matches!(
1314 schema,
1315 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null))
1316 )
1317 };
1318 match (is_null(&variant[0]), is_null(&variant[1])) {
1319 (true, false) => Some((Nullability::NullFirst, &variant[1])),
1320 (false, true) => Some((Nullability::NullSecond, &variant[0])),
1321 _ => None,
1322 }
1323}
1324
1325#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1326enum UnionBranchKey {
1327 Named(String),
1328 Primitive(PrimitiveType),
1329 Array,
1330 Map,
1331}
1332
1333fn branch_key_of<'a>(s: &Schema<'a>, enclosing_ns: Option<&'a str>) -> Option<UnionBranchKey> {
1334 let (name, namespace) = match s {
1335 Schema::TypeName(TypeName::Primitive(p))
1336 | Schema::Type(Type {
1337 r#type: TypeName::Primitive(p),
1338 ..
1339 }) => return Some(UnionBranchKey::Primitive(*p)),
1340 Schema::TypeName(TypeName::Ref(name))
1341 | Schema::Type(Type {
1342 r#type: TypeName::Ref(name),
1343 ..
1344 }) => (name, None),
1345 Schema::Complex(ComplexType::Array(_)) => return Some(UnionBranchKey::Array),
1346 Schema::Complex(ComplexType::Map(_)) => return Some(UnionBranchKey::Map),
1347 Schema::Complex(ComplexType::Record(r)) => (&r.name, r.namespace),
1348 Schema::Complex(ComplexType::Enum(e)) => (&e.name, e.namespace),
1349 Schema::Complex(ComplexType::Fixed(f)) => (&f.name, f.namespace),
1350 Schema::Union(_) => return None,
1351 };
1352 let (full, _) = make_full_name(name, namespace, enclosing_ns);
1353 Some(UnionBranchKey::Named(full))
1354}
1355
1356fn union_first_duplicate<'a>(
1357 branches: &'a [Schema<'a>],
1358 enclosing_ns: Option<&'a str>,
1359) -> Option<String> {
1360 let mut seen = HashSet::with_capacity(branches.len());
1361 for schema in branches {
1362 if let Some(key) = branch_key_of(schema, enclosing_ns)
1363 && !seen.insert(key.clone())
1364 {
1365 let msg = match key {
1366 UnionBranchKey::Named(full) => format!("named type {full}"),
1367 UnionBranchKey::Primitive(p) => format!("primitive {}", p.as_ref()),
1368 UnionBranchKey::Array => "array".to_string(),
1369 UnionBranchKey::Map => "map".to_string(),
1370 };
1371 return Some(msg);
1372 }
1373 }
1374 None
1375}
1376
1377struct Maker<'a> {
1381 resolver: Resolver<'a>,
1382 use_utf8view: bool,
1383 strict_mode: bool,
1384 tz: Tz,
1385}
1386
1387impl<'a> Maker<'a> {
1388 fn new(use_utf8view: bool, strict_mode: bool, tz: Tz) -> Self {
1389 Self {
1390 resolver: Default::default(),
1391 use_utf8view,
1392 strict_mode,
1393 tz,
1394 }
1395 }
1396
1397 #[cfg(feature = "avro_custom_types")]
1398 #[inline]
1399 fn propagate_nullability_into_ree(dt: &mut AvroDataType, nb: Nullability) {
1400 if let Codec::RunEndEncoded(values, bits) = dt.codec.clone() {
1401 let mut inner = (*values).clone();
1402 inner.nullability = Some(nb);
1403 dt.codec = Codec::RunEndEncoded(Arc::new(inner), bits);
1404 }
1405 }
1406
1407 fn make_data_type<'s>(
1408 &mut self,
1409 writer_schema: &'s Schema<'a>,
1410 reader_schema: Option<&'s Schema<'a>>,
1411 namespace: Option<&'a str>,
1412 ) -> Result<AvroDataType, ArrowError> {
1413 match reader_schema {
1414 Some(reader_schema) => self.resolve_type(writer_schema, reader_schema, namespace),
1415 None => self.parse_type(writer_schema, namespace),
1416 }
1417 }
1418
1419 fn parse_type<'s>(
1432 &mut self,
1433 schema: &'s Schema<'a>,
1434 namespace: Option<&'a str>,
1435 ) -> Result<AvroDataType, ArrowError> {
1436 match schema {
1437 Schema::TypeName(TypeName::Primitive(p)) => Ok(AvroDataType::new(
1438 Codec::from(*p).with_utf8view(self.use_utf8view),
1439 Default::default(),
1440 None,
1441 )),
1442 Schema::TypeName(TypeName::Ref(name)) => self.resolver.resolve(name, namespace),
1443 Schema::Union(f) => {
1444 let null = f
1445 .iter()
1446 .position(|x| x == &Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)));
1447 match (f.len() == 2, null) {
1448 (true, Some(0)) => {
1449 let mut field = self.parse_type(&f[1], namespace)?;
1450 field.nullability = Some(Nullability::NullFirst);
1451 #[cfg(feature = "avro_custom_types")]
1452 Self::propagate_nullability_into_ree(&mut field, Nullability::NullFirst);
1453 return Ok(field);
1454 }
1455 (true, Some(1)) => {
1456 if self.strict_mode {
1457 return Err(ArrowError::SchemaError(
1458 "Found Avro union of the form ['T','null'], which is disallowed in strict_mode"
1459 .to_string(),
1460 ));
1461 }
1462 let mut field = self.parse_type(&f[0], namespace)?;
1463 field.nullability = Some(Nullability::NullSecond);
1464 #[cfg(feature = "avro_custom_types")]
1465 Self::propagate_nullability_into_ree(&mut field, Nullability::NullSecond);
1466 return Ok(field);
1467 }
1468 _ => {}
1469 }
1470 if f.iter().any(|s| matches!(s, Schema::Union(_))) {
1472 return Err(ArrowError::SchemaError(
1473 "Avro unions may not immediately contain other unions".to_string(),
1474 ));
1475 }
1476 if let Some(dup) = union_first_duplicate(f, namespace) {
1478 return Err(ArrowError::SchemaError(format!(
1479 "Avro union contains duplicate branch type: {dup}"
1480 )));
1481 }
1482 let children: Vec<AvroDataType> = f
1484 .iter()
1485 .map(|s| self.parse_type(s, namespace))
1486 .collect::<Result<_, _>>()?;
1487 let union_fields = build_union_fields(&children)?;
1489 Ok(AvroDataType::new(
1490 Codec::Union(Arc::from(children), union_fields, UnionMode::Dense),
1491 Default::default(),
1492 None,
1493 ))
1494 }
1495 Schema::Complex(c) => match c {
1496 ComplexType::Record(r) => {
1497 let namespace = r.namespace.or(namespace);
1498 let mut metadata = r.attributes.field_metadata();
1499 let fields = r
1500 .fields
1501 .iter()
1502 .map(|field| {
1503 Ok(AvroField {
1504 name: field.name.to_string(),
1505 data_type: self.parse_type(&field.r#type, namespace)?,
1506 })
1507 })
1508 .collect::<Result<_, ArrowError>>()?;
1509 metadata.insert(AVRO_NAME_METADATA_KEY.to_string(), r.name.to_string());
1510 if let Some(ns) = namespace {
1511 metadata.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
1512 }
1513 let field = AvroDataType {
1514 nullability: None,
1515 codec: Codec::Struct(fields),
1516 metadata,
1517 resolution: None,
1518 };
1519 self.resolver.register(r.name, namespace, field.clone());
1520 Ok(field)
1521 }
1522 ComplexType::Array(a) => {
1523 let field = self.parse_type(a.items.as_ref(), namespace)?;
1524 Ok(AvroDataType {
1525 nullability: None,
1526 metadata: a.attributes.field_metadata(),
1527 codec: Codec::List(Arc::new(field)),
1528 resolution: None,
1529 })
1530 }
1531 ComplexType::Fixed(f) => {
1532 let size = f.size.try_into().map_err(|e| {
1533 ArrowError::ParseError(format!("Overflow converting size to i32: {e}"))
1534 })?;
1535 let namespace = f.namespace.or(namespace);
1536 let mut metadata = f.attributes.field_metadata();
1537 metadata.insert(AVRO_NAME_METADATA_KEY.to_string(), f.name.to_string());
1538 if let Some(ns) = namespace {
1539 metadata.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
1540 }
1541 let field = match f.attributes.logical_type {
1542 Some("decimal") => {
1543 let (precision, scale, _) =
1544 parse_decimal_attributes(&f.attributes, Some(size as usize), true)?;
1545 AvroDataType {
1546 nullability: None,
1547 metadata,
1548 codec: Codec::Decimal(precision, Some(scale), Some(size as usize)),
1549 resolution: None,
1550 }
1551 }
1552 Some("duration") => {
1553 if size != 12 {
1554 return Err(ArrowError::ParseError(format!(
1555 "Invalid fixed size for Duration: {size}, must be 12"
1556 )));
1557 }
1558 AvroDataType {
1559 nullability: None,
1560 metadata,
1561 codec: Codec::Interval,
1562 resolution: None,
1563 }
1564 }
1565 Some("uuid") => {
1566 if size != 16 {
1567 return Err(ArrowError::ParseError(format!(
1568 "Invalid fixed size for UUID: {size}, must be 16"
1569 )));
1570 }
1571 metadata.insert("logicalType".into(), "uuid".into());
1572 AvroDataType {
1573 nullability: None,
1574 metadata,
1575 codec: Codec::Fixed(size),
1576 resolution: None,
1577 }
1578 }
1579 #[cfg(feature = "avro_custom_types")]
1580 Some("arrow.uint64") if size == 8 => AvroDataType {
1581 nullability: None,
1582 metadata,
1583 codec: Codec::UInt64,
1584 resolution: None,
1585 },
1586 #[cfg(feature = "avro_custom_types")]
1587 Some("arrow.float16") if size == 2 => AvroDataType {
1588 nullability: None,
1589 metadata,
1590 codec: Codec::Float16,
1591 resolution: None,
1592 },
1593 #[cfg(feature = "avro_custom_types")]
1594 Some("arrow.interval-year-month") if size == 4 => AvroDataType {
1595 nullability: None,
1596 metadata,
1597 codec: Codec::IntervalYearMonth,
1598 resolution: None,
1599 },
1600 #[cfg(feature = "avro_custom_types")]
1601 Some("arrow.interval-month-day-nano") if size == 16 => AvroDataType {
1602 nullability: None,
1603 metadata,
1604 codec: Codec::IntervalMonthDayNano,
1605 resolution: None,
1606 },
1607 #[cfg(feature = "avro_custom_types")]
1608 Some("arrow.interval-day-time") if size == 8 => AvroDataType {
1609 nullability: None,
1610 metadata,
1611 codec: Codec::IntervalDayTime,
1612 resolution: None,
1613 },
1614 _ => AvroDataType {
1615 nullability: None,
1616 metadata,
1617 codec: Codec::Fixed(size),
1618 resolution: None,
1619 },
1620 };
1621 self.resolver.register(f.name, namespace, field.clone());
1622 Ok(field)
1623 }
1624 ComplexType::Enum(e) => {
1625 let namespace = e.namespace.or(namespace);
1626 let symbols = e
1627 .symbols
1628 .iter()
1629 .map(|s| s.to_string())
1630 .collect::<Arc<[String]>>();
1631 let mut metadata = e.attributes.field_metadata();
1632 let symbols_json = serde_json::to_string(&e.symbols).map_err(|e| {
1633 ArrowError::ParseError(format!("Failed to serialize enum symbols: {e}"))
1634 })?;
1635 metadata.insert(AVRO_ENUM_SYMBOLS_METADATA_KEY.to_string(), symbols_json);
1636 metadata.insert(AVRO_NAME_METADATA_KEY.to_string(), e.name.to_string());
1637 if let Some(ns) = namespace {
1638 metadata.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
1639 }
1640 let field = AvroDataType {
1641 nullability: None,
1642 metadata,
1643 codec: Codec::Enum(symbols),
1644 resolution: None,
1645 };
1646 self.resolver.register(e.name, namespace, field.clone());
1647 Ok(field)
1648 }
1649 ComplexType::Map(m) => {
1650 let val = self.parse_type(&m.values, namespace)?;
1651 Ok(AvroDataType {
1652 nullability: None,
1653 metadata: m.attributes.field_metadata(),
1654 codec: Codec::Map(Arc::new(val)),
1655 resolution: None,
1656 })
1657 }
1658 },
1659 Schema::Type(t) => {
1660 let mut field = self.parse_type(&Schema::TypeName(t.r#type.clone()), namespace)?;
1661 match (t.attributes.logical_type, &mut field.codec) {
1663 (Some("decimal"), c @ Codec::Binary) => {
1664 let (prec, sc, _) = parse_decimal_attributes(&t.attributes, None, false)?;
1665 *c = Codec::Decimal(prec, Some(sc), None);
1666 }
1667 (Some("date"), c @ Codec::Int32) => *c = Codec::Date32,
1668 (Some("time-millis"), c @ Codec::Int32) => *c = Codec::TimeMillis,
1669 (Some("time-micros"), c @ Codec::Int64) => *c = Codec::TimeMicros,
1670 (Some("timestamp-millis"), c @ Codec::Int64) => {
1671 *c = Codec::TimestampMillis(Some(self.tz))
1672 }
1673 (Some("timestamp-micros"), c @ Codec::Int64) => {
1674 *c = Codec::TimestampMicros(Some(self.tz))
1675 }
1676 (Some("local-timestamp-millis"), c @ Codec::Int64) => {
1677 *c = Codec::TimestampMillis(None)
1678 }
1679 (Some("local-timestamp-micros"), c @ Codec::Int64) => {
1680 *c = Codec::TimestampMicros(None)
1681 }
1682 (Some("timestamp-nanos"), c @ Codec::Int64) => {
1683 *c = Codec::TimestampNanos(Some(self.tz))
1684 }
1685 (Some("local-timestamp-nanos"), c @ Codec::Int64) => {
1686 *c = Codec::TimestampNanos(None)
1687 }
1688 (Some("uuid"), c @ Codec::Utf8) => {
1689 *c = Codec::Uuid;
1693 field.metadata.insert("logicalType".into(), "uuid".into());
1694 }
1695 #[cfg(feature = "avro_custom_types")]
1696 (Some("arrow.duration-nanos"), c @ Codec::Int64) => *c = Codec::DurationNanos,
1697 #[cfg(feature = "avro_custom_types")]
1698 (Some("arrow.duration-micros"), c @ Codec::Int64) => *c = Codec::DurationMicros,
1699 #[cfg(feature = "avro_custom_types")]
1700 (Some("arrow.duration-millis"), c @ Codec::Int64) => *c = Codec::DurationMillis,
1701 #[cfg(feature = "avro_custom_types")]
1702 (Some("arrow.duration-seconds"), c @ Codec::Int64) => {
1703 *c = Codec::DurationSeconds
1704 }
1705 #[cfg(feature = "avro_custom_types")]
1706 (Some("arrow.run-end-encoded"), _) => {
1707 let bits_u8: u8 = t
1708 .attributes
1709 .additional
1710 .get("arrow.runEndIndexBits")
1711 .and_then(|v| v.as_u64())
1712 .and_then(|n| u8::try_from(n).ok())
1713 .ok_or_else(|| ArrowError::ParseError(
1714 "arrow.run-end-encoded requires 'arrow.runEndIndexBits' (one of 16, 32, or 64)"
1715 .to_string(),
1716 ))?;
1717 if bits_u8 != 16 && bits_u8 != 32 && bits_u8 != 64 {
1718 return Err(ArrowError::ParseError(format!(
1719 "Invalid 'arrow.runEndIndexBits' value {bits_u8}; must be 16, 32, or 64"
1720 )));
1721 }
1722 let values_site = field.clone();
1724 field.codec = Codec::RunEndEncoded(Arc::new(values_site), bits_u8);
1725 }
1726 #[cfg(feature = "avro_custom_types")]
1728 (Some("arrow.int8"), c @ Codec::Int32) => *c = Codec::Int8,
1729 #[cfg(feature = "avro_custom_types")]
1730 (Some("arrow.int16"), c @ Codec::Int32) => *c = Codec::Int16,
1731 #[cfg(feature = "avro_custom_types")]
1732 (Some("arrow.uint8"), c @ Codec::Int32) => *c = Codec::UInt8,
1733 #[cfg(feature = "avro_custom_types")]
1734 (Some("arrow.uint16"), c @ Codec::Int32) => *c = Codec::UInt16,
1735 #[cfg(feature = "avro_custom_types")]
1736 (Some("arrow.uint32"), c @ Codec::Int64) => *c = Codec::UInt32,
1737 #[cfg(feature = "avro_custom_types")]
1738 (Some("arrow.uint64"), c @ Codec::Fixed(8)) => *c = Codec::UInt64,
1739 #[cfg(feature = "avro_custom_types")]
1741 (Some("arrow.float16"), c @ Codec::Fixed(2)) => *c = Codec::Float16,
1742 #[cfg(feature = "avro_custom_types")]
1744 (Some("arrow.date64"), c @ Codec::Int64) => *c = Codec::Date64,
1745 #[cfg(feature = "avro_custom_types")]
1747 (Some("arrow.time64-nanosecond"), c @ Codec::Int64) => *c = Codec::TimeNanos,
1748 #[cfg(feature = "avro_custom_types")]
1749 (Some("arrow.time32-second"), c @ Codec::Int32) => *c = Codec::Time32Secs,
1750 #[cfg(feature = "avro_custom_types")]
1751 (Some("arrow.timestamp-second"), c @ Codec::Int64) => {
1752 *c = Codec::TimestampSecs(true)
1753 }
1754 #[cfg(feature = "avro_custom_types")]
1755 (Some("arrow.local-timestamp-second"), c @ Codec::Int64) => {
1756 *c = Codec::TimestampSecs(false)
1757 }
1758 #[cfg(feature = "avro_custom_types")]
1760 (Some("arrow.interval-year-month"), c @ Codec::Fixed(4)) => {
1761 *c = Codec::IntervalYearMonth
1762 }
1763 #[cfg(feature = "avro_custom_types")]
1764 (Some("arrow.interval-month-day-nano"), c @ Codec::Fixed(16)) => {
1765 *c = Codec::IntervalMonthDayNano
1766 }
1767 #[cfg(feature = "avro_custom_types")]
1768 (Some("arrow.interval-day-time"), c @ Codec::Fixed(8)) => {
1769 *c = Codec::IntervalDayTime
1770 }
1771 (Some(logical), _) => {
1772 field.metadata.insert("logicalType".into(), logical.into());
1774 }
1775 (None, _) => {}
1776 }
1777 if matches!(field.codec, Codec::Int64)
1778 && let Some(unit) = t
1779 .attributes
1780 .additional
1781 .get("arrowTimeUnit")
1782 .and_then(|v| v.as_str())
1783 && unit == "nanosecond"
1784 {
1785 field.codec = Codec::TimestampNanos(Some(self.tz));
1786 }
1787 if !t.attributes.additional.is_empty() {
1788 for (k, v) in &t.attributes.additional {
1789 field.metadata.insert(k.to_string(), v.to_string());
1790 }
1791 }
1792 Ok(field)
1793 }
1794 }
1795 }
1796
1797 fn resolve_type<'s>(
1798 &mut self,
1799 writer_schema: &'s Schema<'a>,
1800 reader_schema: &'s Schema<'a>,
1801 namespace: Option<&'a str>,
1802 ) -> Result<AvroDataType, ArrowError> {
1803 if let (Some(write_primitive), Some(read_primitive)) =
1804 (primitive_of(writer_schema), primitive_of(reader_schema))
1805 {
1806 return self.resolve_primitives(write_primitive, read_primitive, reader_schema);
1807 }
1808 match (writer_schema, reader_schema) {
1809 (Schema::Union(writer_variants), Schema::Union(reader_variants)) => {
1810 let writer_variants = writer_variants.as_slice();
1811 let reader_variants = reader_variants.as_slice();
1812 match (
1813 nullable_union_variants(writer_variants),
1814 nullable_union_variants(reader_variants),
1815 ) {
1816 (Some((w_nb, w_nonnull)), Some((r_nb, r_nonnull))) => {
1817 let mut dt = self.resolve_type(w_nonnull, r_nonnull, namespace)?;
1818 let mut writer_to_reader = vec![None, None];
1819 writer_to_reader[w_nb.non_null_index()] = Some((
1820 r_nb.non_null_index(),
1821 dt.resolution
1822 .take()
1823 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct)),
1824 ));
1825 dt.nullability = Some(w_nb);
1826 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
1827 writer_to_reader: Arc::from(writer_to_reader),
1828 writer_is_union: true,
1829 reader_is_union: true,
1830 }));
1831 #[cfg(feature = "avro_custom_types")]
1832 Self::propagate_nullability_into_ree(&mut dt, w_nb);
1833 Ok(dt)
1834 }
1835 _ => self.resolve_unions(writer_variants, reader_variants, namespace),
1836 }
1837 }
1838 (Schema::Union(writer_variants), reader_non_union) => {
1839 let writer_to_reader: Vec<Option<(usize, ResolutionInfo)>> = writer_variants
1840 .iter()
1841 .map(|writer| {
1842 self.resolve_type(writer, reader_non_union, namespace)
1843 .ok()
1844 .map(|tmp| {
1845 let resolution = tmp
1846 .resolution
1847 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct));
1848 (0usize, resolution)
1849 })
1850 })
1851 .collect();
1852 let mut dt = self.parse_type(reader_non_union, namespace)?;
1853 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
1854 writer_to_reader: Arc::from(writer_to_reader),
1855 writer_is_union: true,
1856 reader_is_union: false,
1857 }));
1858 Ok(dt)
1859 }
1860 (writer_non_union, Schema::Union(reader_variants)) => {
1861 if let Some((nullability, non_null_branch)) =
1862 nullable_union_variants(reader_variants)
1863 {
1864 let mut dt = self.resolve_type(writer_non_union, non_null_branch, namespace)?;
1865 #[cfg(feature = "avro_custom_types")]
1866 Self::propagate_nullability_into_ree(&mut dt, nullability);
1867 dt.nullability = Some(nullability);
1868 if dt.resolution.is_none() {
1871 dt.resolution = Some(ResolutionInfo::Promotion(Promotion::Direct));
1872 }
1873 Ok(dt)
1874 } else {
1875 let Some((match_idx, mut match_dt)) =
1876 self.find_best_union_match(writer_non_union, reader_variants, namespace)
1877 else {
1878 return Err(ArrowError::SchemaError(
1879 "Writer schema does not match any reader union branch".to_string(),
1880 ));
1881 };
1882 let resolution = match_dt
1888 .resolution
1889 .take()
1890 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct));
1891 let mut match_dt = Some(match_dt);
1892 let children = reader_variants
1893 .iter()
1894 .enumerate()
1895 .map(|(idx, variant)| {
1896 if idx == match_idx {
1897 Ok(match_dt.take().unwrap())
1898 } else {
1899 self.parse_type(variant, namespace)
1900 }
1901 })
1902 .collect::<Result<Vec<_>, _>>()?;
1903 let union_fields = build_union_fields(&children)?;
1904 let mut dt = AvroDataType::new(
1905 Codec::Union(children.into(), union_fields, UnionMode::Dense),
1906 Default::default(),
1907 None,
1908 );
1909 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
1910 writer_to_reader: Arc::from(vec![Some((match_idx, resolution))]),
1911 writer_is_union: false,
1912 reader_is_union: true,
1913 }));
1914 Ok(dt)
1915 }
1916 }
1917 (
1918 Schema::Complex(ComplexType::Array(writer_array)),
1919 Schema::Complex(ComplexType::Array(reader_array)),
1920 ) => self.resolve_array(writer_array, reader_array, namespace),
1921 (
1922 Schema::Complex(ComplexType::Map(writer_map)),
1923 Schema::Complex(ComplexType::Map(reader_map)),
1924 ) => self.resolve_map(writer_map, reader_map, namespace),
1925 (
1926 Schema::Complex(ComplexType::Fixed(writer_fixed)),
1927 Schema::Complex(ComplexType::Fixed(reader_fixed)),
1928 ) => self.resolve_fixed(writer_fixed, reader_fixed, reader_schema, namespace),
1929 (
1930 Schema::Complex(ComplexType::Record(writer_record)),
1931 Schema::Complex(ComplexType::Record(reader_record)),
1932 ) => self.resolve_records(writer_record, reader_record, namespace),
1933 (
1934 Schema::Complex(ComplexType::Enum(writer_enum)),
1935 Schema::Complex(ComplexType::Enum(reader_enum)),
1936 ) => self.resolve_enums(writer_enum, reader_enum, reader_schema, namespace),
1937 (Schema::TypeName(TypeName::Ref(_)), _) => self.parse_type(reader_schema, namespace),
1938 (_, Schema::TypeName(TypeName::Ref(_))) => self.parse_type(reader_schema, namespace),
1939 _ => Err(ArrowError::NotYetImplemented(
1940 "Other resolutions not yet implemented".to_string(),
1941 )),
1942 }
1943 }
1944
1945 fn find_best_union_match(
1946 &mut self,
1947 writer: &Schema<'a>,
1948 reader_variants: &[Schema<'a>],
1949 namespace: Option<&'a str>,
1950 ) -> Option<(usize, AvroDataType)> {
1951 let mut first_resolution = None;
1952 for (reader_index, reader) in reader_variants.iter().enumerate() {
1953 if let Ok(dt) = self.resolve_type(writer, reader, namespace) {
1954 match &dt.resolution {
1955 None | Some(ResolutionInfo::Promotion(Promotion::Direct)) => {
1956 return Some((reader_index, dt));
1958 }
1959 Some(_) => {
1960 if first_resolution.is_none() {
1961 first_resolution = Some((reader_index, dt));
1963 }
1964 }
1965 }
1966 }
1967 }
1968 first_resolution
1969 }
1970
1971 fn resolve_unions<'s>(
1972 &mut self,
1973 writer_variants: &'s [Schema<'a>],
1974 reader_variants: &'s [Schema<'a>],
1975 namespace: Option<&'a str>,
1976 ) -> Result<AvroDataType, ArrowError> {
1977 let mut resolved_reader_encodings = HashMap::new();
1978 let writer_to_reader: Vec<Option<(usize, ResolutionInfo)>> = writer_variants
1979 .iter()
1980 .map(|writer| {
1981 self.find_best_union_match(writer, reader_variants, namespace)
1982 .map(|(match_idx, mut match_dt)| {
1983 let resolution = match_dt
1984 .resolution
1985 .take()
1986 .unwrap_or(ResolutionInfo::Promotion(Promotion::Direct));
1987 resolved_reader_encodings.insert(match_idx, match_dt);
1990 (match_idx, resolution)
1991 })
1992 })
1993 .collect();
1994 let reader_encodings: Vec<AvroDataType> = reader_variants
1995 .iter()
1996 .enumerate()
1997 .map(|(reader_idx, reader_schema)| {
1998 if let Some(resolved) = resolved_reader_encodings.remove(&reader_idx) {
1999 Ok(resolved)
2000 } else {
2001 self.parse_type(reader_schema, namespace)
2002 }
2003 })
2004 .collect::<Result<_, _>>()?;
2005 let union_fields = build_union_fields(&reader_encodings)?;
2006 let mut dt = AvroDataType::new(
2007 Codec::Union(reader_encodings.into(), union_fields, UnionMode::Dense),
2008 Default::default(),
2009 None,
2010 );
2011 dt.resolution = Some(ResolutionInfo::Union(ResolvedUnion {
2012 writer_to_reader: Arc::from(writer_to_reader),
2013 writer_is_union: true,
2014 reader_is_union: true,
2015 }));
2016 Ok(dt)
2017 }
2018
2019 fn resolve_array(
2020 &mut self,
2021 writer_array: &Array<'a>,
2022 reader_array: &Array<'a>,
2023 namespace: Option<&'a str>,
2024 ) -> Result<AvroDataType, ArrowError> {
2025 Ok(AvroDataType {
2026 nullability: None,
2027 metadata: reader_array.attributes.field_metadata(),
2028 codec: Codec::List(Arc::new(self.make_data_type(
2029 writer_array.items.as_ref(),
2030 Some(reader_array.items.as_ref()),
2031 namespace,
2032 )?)),
2033 resolution: None,
2034 })
2035 }
2036
2037 fn resolve_map(
2038 &mut self,
2039 writer_map: &Map<'a>,
2040 reader_map: &Map<'a>,
2041 namespace: Option<&'a str>,
2042 ) -> Result<AvroDataType, ArrowError> {
2043 Ok(AvroDataType {
2044 nullability: None,
2045 metadata: reader_map.attributes.field_metadata(),
2046 codec: Codec::Map(Arc::new(self.make_data_type(
2047 &writer_map.values,
2048 Some(&reader_map.values),
2049 namespace,
2050 )?)),
2051 resolution: None,
2052 })
2053 }
2054
2055 fn resolve_fixed<'s>(
2056 &mut self,
2057 writer_fixed: &Fixed<'a>,
2058 reader_fixed: &Fixed<'a>,
2059 reader_schema: &'s Schema<'a>,
2060 namespace: Option<&'a str>,
2061 ) -> Result<AvroDataType, ArrowError> {
2062 ensure_names_match(
2063 "Fixed",
2064 writer_fixed.name,
2065 writer_fixed.namespace,
2066 &writer_fixed.aliases,
2067 reader_fixed.name,
2068 reader_fixed.namespace,
2069 &reader_fixed.aliases,
2070 )?;
2071 if writer_fixed.size != reader_fixed.size {
2072 return Err(ArrowError::SchemaError(format!(
2073 "Fixed size mismatch for {}: writer={}, reader={}",
2074 reader_fixed.name, writer_fixed.size, reader_fixed.size
2075 )));
2076 }
2077 self.parse_type(reader_schema, namespace)
2078 }
2079
2080 fn resolve_primitives(
2081 &mut self,
2082 write_primitive: PrimitiveType,
2083 read_primitive: PrimitiveType,
2084 reader_schema: &Schema<'a>,
2085 ) -> Result<AvroDataType, ArrowError> {
2086 if write_primitive == read_primitive {
2087 return self.parse_type(reader_schema, None);
2088 }
2089 let promotion = match (write_primitive, read_primitive) {
2090 (PrimitiveType::Int, PrimitiveType::Long) => Promotion::IntToLong,
2091 (PrimitiveType::Int, PrimitiveType::Float) => Promotion::IntToFloat,
2092 (PrimitiveType::Int, PrimitiveType::Double) => Promotion::IntToDouble,
2093 (PrimitiveType::Long, PrimitiveType::Float) => Promotion::LongToFloat,
2094 (PrimitiveType::Long, PrimitiveType::Double) => Promotion::LongToDouble,
2095 (PrimitiveType::Float, PrimitiveType::Double) => Promotion::FloatToDouble,
2096 (PrimitiveType::String, PrimitiveType::Bytes) => Promotion::StringToBytes,
2097 (PrimitiveType::Bytes, PrimitiveType::String) => Promotion::BytesToString,
2098 _ => {
2099 return Err(ArrowError::ParseError(format!(
2100 "Illegal promotion {write_primitive:?} to {read_primitive:?}"
2101 )));
2102 }
2103 };
2104 let mut datatype = self.parse_type(reader_schema, None)?;
2105 datatype.resolution = Some(ResolutionInfo::Promotion(promotion));
2106 Ok(datatype)
2107 }
2108
2109 fn resolve_enums(
2165 &mut self,
2166 writer_enum: &Enum<'a>,
2167 reader_enum: &Enum<'a>,
2168 reader_schema: &Schema<'a>,
2169 namespace: Option<&'a str>,
2170 ) -> Result<AvroDataType, ArrowError> {
2171 ensure_names_match(
2172 "Enum",
2173 writer_enum.name,
2174 writer_enum.namespace,
2175 &writer_enum.aliases,
2176 reader_enum.name,
2177 reader_enum.namespace,
2178 &reader_enum.aliases,
2179 )?;
2180 if writer_enum.symbols == reader_enum.symbols {
2181 return self.parse_type(reader_schema, namespace);
2182 }
2183 let reader_index: HashMap<&str, i32> = reader_enum
2184 .symbols
2185 .iter()
2186 .enumerate()
2187 .map(|(index, &symbol)| (symbol, index as i32))
2188 .collect();
2189 let default_index: i32 = match reader_enum.default {
2190 Some(symbol) => *reader_index.get(symbol).ok_or_else(|| {
2191 ArrowError::SchemaError(format!(
2192 "Reader enum '{}' default symbol '{symbol}' not found in symbols list",
2193 reader_enum.name,
2194 ))
2195 })?,
2196 None => -1,
2197 };
2198 let mapping: Vec<i32> = writer_enum
2199 .symbols
2200 .iter()
2201 .map(|&write_symbol| {
2202 reader_index
2203 .get(write_symbol)
2204 .copied()
2205 .unwrap_or(default_index)
2206 })
2207 .collect();
2208 if self.strict_mode && mapping.iter().any(|&m| m < 0) {
2209 return Err(ArrowError::SchemaError(format!(
2210 "Reader enum '{}' does not cover all writer symbols and no default is provided",
2211 reader_enum.name
2212 )));
2213 }
2214 let mut dt = self.parse_type(reader_schema, namespace)?;
2215 dt.resolution = Some(ResolutionInfo::EnumMapping(EnumMapping {
2216 mapping: Arc::from(mapping),
2217 default_index,
2218 }));
2219 let reader_ns = reader_enum.namespace.or(namespace);
2220 self.resolver
2221 .register(reader_enum.name, reader_ns, dt.clone());
2222 Ok(dt)
2223 }
2224
2225 #[inline]
2226 fn build_writer_lookup(
2227 writer_record: &Record<'a>,
2228 ) -> (HashMap<&'a str, usize>, HashSet<&'a str>) {
2229 let mut map: HashMap<&str, usize> = HashMap::with_capacity(writer_record.fields.len() * 2);
2230 for (idx, wf) in writer_record.fields.iter().enumerate() {
2231 map.insert(wf.name, idx);
2233 }
2234 let mut ambiguous: HashSet<&str> = HashSet::new();
2236 for (idx, wf) in writer_record.fields.iter().enumerate() {
2237 for &alias in &wf.aliases {
2238 match map.entry(alias) {
2239 Entry::Occupied(e) if *e.get() != idx => {
2240 ambiguous.insert(alias);
2241 }
2242 Entry::Vacant(e) => {
2243 e.insert(idx);
2244 }
2245 Entry::Occupied(_) => {}
2246 }
2247 }
2248 }
2249 (map, ambiguous)
2250 }
2251
2252 fn resolve_records(
2253 &mut self,
2254 writer_record: &Record<'a>,
2255 reader_record: &Record<'a>,
2256 namespace: Option<&'a str>,
2257 ) -> Result<AvroDataType, ArrowError> {
2258 ensure_names_match(
2259 "Record",
2260 writer_record.name,
2261 writer_record.namespace,
2262 &writer_record.aliases,
2263 reader_record.name,
2264 reader_record.namespace,
2265 &reader_record.aliases,
2266 )?;
2267 let writer_ns = writer_record.namespace.or(namespace);
2268 let reader_ns = reader_record.namespace.or(namespace);
2269 let mut reader_md = reader_record.attributes.field_metadata();
2270 reader_md.insert(
2271 AVRO_NAME_METADATA_KEY.to_string(),
2272 reader_record.name.to_string(),
2273 );
2274 if let Some(ns) = reader_ns {
2275 reader_md.insert(AVRO_NAMESPACE_METADATA_KEY.to_string(), ns.to_string());
2276 }
2277 let (writer_lookup, ambiguous_writer_aliases) = Self::build_writer_lookup(writer_record);
2279 let mut writer_to_reader: Vec<Option<usize>> = vec![None; writer_record.fields.len()];
2280 let mut reader_fields: Vec<AvroField> = Vec::with_capacity(reader_record.fields.len());
2281 let mut default_fields: Vec<usize> = Vec::new();
2283 for (reader_idx, r_field) in reader_record.fields.iter().enumerate() {
2284 let mut match_idx = writer_lookup.get(r_field.name).copied();
2286 let mut matched_via_alias: Option<&str> = None;
2287 if match_idx.is_none() {
2288 for &alias in &r_field.aliases {
2289 if let Some(i) = writer_lookup.get(alias).copied() {
2290 if self.strict_mode && ambiguous_writer_aliases.contains(alias) {
2291 return Err(ArrowError::SchemaError(format!(
2292 "Ambiguous alias '{alias}' on reader field '{}' matches multiple writer fields",
2293 r_field.name
2294 )));
2295 }
2296 match_idx = Some(i);
2297 matched_via_alias = Some(alias);
2298 break;
2299 }
2300 }
2301 }
2302 if let Some(wi) = match_idx {
2303 if writer_to_reader[wi].is_none() {
2304 let w_schema = &writer_record.fields[wi].r#type;
2305 let dt = self.make_data_type(w_schema, Some(&r_field.r#type), reader_ns)?;
2306 writer_to_reader[wi] = Some(reader_idx);
2307 reader_fields.push(AvroField {
2308 name: r_field.name.to_owned(),
2309 data_type: dt,
2310 });
2311 continue;
2312 } else if self.strict_mode {
2313 let existing_reader = writer_to_reader[wi].unwrap();
2315 let via = matched_via_alias
2316 .map(|a| format!("alias '{a}'"))
2317 .unwrap_or_else(|| "name match".to_string());
2318 return Err(ArrowError::SchemaError(format!(
2319 "Multiple reader fields map to the same writer field '{}' via {via} (existing reader index {existing_reader}, new reader index {reader_idx})",
2320 writer_record.fields[wi].name
2321 )));
2322 }
2323 }
2325 let mut dt = self.parse_type(&r_field.r#type, reader_ns)?;
2327 if let Some(default_json) = r_field.default.as_ref() {
2328 dt.resolution = Some(ResolutionInfo::DefaultValue(
2329 dt.parse_and_store_default(default_json)?,
2330 ));
2331 default_fields.push(reader_idx);
2332 } else if dt.nullability() == Some(Nullability::NullFirst) {
2333 dt.resolution = Some(ResolutionInfo::DefaultValue(
2335 dt.parse_and_store_default(&Value::Null)?,
2336 ));
2337 default_fields.push(reader_idx);
2338 } else {
2339 return Err(ArrowError::SchemaError(format!(
2340 "Reader field '{}' not present in writer schema must have a default value",
2341 r_field.name
2342 )));
2343 }
2344 reader_fields.push(AvroField {
2345 name: r_field.name.to_owned(),
2346 data_type: dt,
2347 });
2348 }
2349 let writer_fields = writer_record
2351 .fields
2352 .iter()
2353 .enumerate()
2354 .map(|(writer_index, writer_field)| {
2355 let dt = self.parse_type(&writer_field.r#type, writer_ns)?;
2356 if let Some(reader_index) = writer_to_reader[writer_index] {
2357 Ok(ResolvedField::ToReader(reader_index, dt))
2358 } else {
2359 Ok(ResolvedField::Skip(dt))
2360 }
2361 })
2362 .collect::<Result<_, ArrowError>>()?;
2363 let resolved = AvroDataType::new_with_resolution(
2364 Codec::Struct(Arc::from(reader_fields)),
2365 reader_md,
2366 None,
2367 Some(ResolutionInfo::Record(ResolvedRecord {
2368 writer_fields,
2369 default_fields: Arc::from(default_fields),
2370 })),
2371 );
2372 self.resolver
2374 .register(reader_record.name, reader_ns, resolved.clone());
2375 Ok(resolved)
2376 }
2377}
2378
2379#[cfg(test)]
2380mod tests {
2381 use super::*;
2382 use crate::schema::{
2383 AVRO_ROOT_RECORD_DEFAULT_NAME, Array, Attributes, ComplexType, Field as AvroFieldSchema,
2384 Fixed, PrimitiveType, Record, Schema, Type, TypeName,
2385 };
2386 use indexmap::IndexMap;
2387 use serde_json::{self, Value};
2388
2389 fn create_schema_with_logical_type(
2390 primitive_type: PrimitiveType,
2391 logical_type: &'static str,
2392 ) -> Schema<'static> {
2393 let attributes = Attributes {
2394 logical_type: Some(logical_type),
2395 additional: Default::default(),
2396 };
2397
2398 Schema::Type(Type {
2399 r#type: TypeName::Primitive(primitive_type),
2400 attributes,
2401 })
2402 }
2403
2404 fn resolve_promotion(writer: PrimitiveType, reader: PrimitiveType) -> AvroDataType {
2405 let writer_schema = Schema::TypeName(TypeName::Primitive(writer));
2406 let reader_schema = Schema::TypeName(TypeName::Primitive(reader));
2407 let mut maker = Maker::new(false, false, Tz::default());
2408 maker
2409 .make_data_type(&writer_schema, Some(&reader_schema), None)
2410 .expect("promotion should resolve")
2411 }
2412
2413 fn mk_primitive(pt: PrimitiveType) -> Schema<'static> {
2414 Schema::TypeName(TypeName::Primitive(pt))
2415 }
2416 fn mk_union(branches: Vec<Schema<'_>>) -> Schema<'_> {
2417 Schema::Union(branches)
2418 }
2419
2420 #[test]
2421 fn test_date_logical_type() {
2422 let schema = create_schema_with_logical_type(PrimitiveType::Int, "date");
2423
2424 let mut maker = Maker::new(false, false, Tz::default());
2425 let result = maker.make_data_type(&schema, None, None).unwrap();
2426
2427 assert!(matches!(result.codec, Codec::Date32));
2428 }
2429
2430 #[test]
2431 fn test_time_millis_logical_type() {
2432 let schema = create_schema_with_logical_type(PrimitiveType::Int, "time-millis");
2433
2434 let mut maker = Maker::new(false, false, Tz::default());
2435 let result = maker.make_data_type(&schema, None, None).unwrap();
2436
2437 assert!(matches!(result.codec, Codec::TimeMillis));
2438 }
2439
2440 #[test]
2441 fn test_time_micros_logical_type() {
2442 let schema = create_schema_with_logical_type(PrimitiveType::Long, "time-micros");
2443
2444 let mut maker = Maker::new(false, false, Tz::default());
2445 let result = maker.make_data_type(&schema, None, None).unwrap();
2446
2447 assert!(matches!(result.codec, Codec::TimeMicros));
2448 }
2449
2450 #[test]
2451 fn test_timestamp_millis_logical_type() {
2452 for tz in [Tz::OffsetZero, Tz::Utc] {
2453 let schema = create_schema_with_logical_type(PrimitiveType::Long, "timestamp-millis");
2454
2455 let mut maker = Maker::new(false, false, tz);
2456 let result = maker.make_data_type(&schema, None, None).unwrap();
2457
2458 let Codec::TimestampMillis(Some(actual_tz)) = result.codec else {
2459 panic!("Expected TimestampMillis codec");
2460 };
2461 assert_eq!(actual_tz, tz);
2462 }
2463 }
2464
2465 #[test]
2466 fn test_timestamp_micros_logical_type() {
2467 for tz in [Tz::OffsetZero, Tz::Utc] {
2468 let schema = create_schema_with_logical_type(PrimitiveType::Long, "timestamp-micros");
2469
2470 let mut maker = Maker::new(false, false, tz);
2471 let result = maker.make_data_type(&schema, None, None).unwrap();
2472
2473 let Codec::TimestampMicros(Some(actual_tz)) = result.codec else {
2474 panic!("Expected TimestampMicros codec");
2475 };
2476 assert_eq!(actual_tz, tz);
2477 }
2478 }
2479
2480 #[test]
2481 fn test_timestamp_nanos_logical_type() {
2482 for tz in [Tz::OffsetZero, Tz::Utc] {
2483 let schema = create_schema_with_logical_type(PrimitiveType::Long, "timestamp-nanos");
2484
2485 let mut maker = Maker::new(false, false, tz);
2486 let result = maker.make_data_type(&schema, None, None).unwrap();
2487
2488 let Codec::TimestampNanos(Some(actual_tz)) = result.codec else {
2489 panic!("Expected TimestampNanos codec");
2490 };
2491 assert_eq!(actual_tz, tz);
2492 }
2493 }
2494
2495 #[test]
2496 fn test_local_timestamp_millis_logical_type() {
2497 let schema = create_schema_with_logical_type(PrimitiveType::Long, "local-timestamp-millis");
2498
2499 let mut maker = Maker::new(false, false, Tz::default());
2500 let result = maker.make_data_type(&schema, None, None).unwrap();
2501
2502 assert!(matches!(result.codec, Codec::TimestampMillis(None)));
2503 }
2504
2505 #[test]
2506 fn test_local_timestamp_micros_logical_type() {
2507 let schema = create_schema_with_logical_type(PrimitiveType::Long, "local-timestamp-micros");
2508
2509 let mut maker = Maker::new(false, false, Tz::default());
2510 let result = maker.make_data_type(&schema, None, None).unwrap();
2511
2512 assert!(matches!(result.codec, Codec::TimestampMicros(None)));
2513 }
2514
2515 #[test]
2516 fn test_local_timestamp_nanos_logical_type() {
2517 let schema = create_schema_with_logical_type(PrimitiveType::Long, "local-timestamp-nanos");
2518
2519 let mut maker = Maker::new(false, false, Tz::default());
2520 let result = maker.make_data_type(&schema, None, None).unwrap();
2521
2522 assert!(matches!(result.codec, Codec::TimestampNanos(None)));
2523 }
2524
2525 #[test]
2526 fn test_uuid_type() {
2527 let mut codec = Codec::Fixed(16);
2528 if let c @ Codec::Fixed(16) = &mut codec {
2529 *c = Codec::Uuid;
2530 }
2531 assert!(matches!(codec, Codec::Uuid));
2532 }
2533
2534 #[test]
2535 fn test_fixed_uuid_logical_type_metadata() {
2536 let schema = Schema::Complex(ComplexType::Fixed(Fixed {
2540 name: "uuid_fixed",
2541 namespace: None,
2542 aliases: vec![],
2543 size: 16,
2544 attributes: Attributes {
2545 logical_type: Some("uuid"),
2546 additional: Default::default(),
2547 },
2548 }));
2549
2550 let mut maker = Maker::new(false, false, Tz::default());
2551 let result = maker.make_data_type(&schema, None, None).unwrap();
2552
2553 assert!(
2554 matches!(result.codec, Codec::Fixed(16)),
2555 "codec should be Fixed(16), got {:?}",
2556 result.codec
2557 );
2558 assert_eq!(
2559 result.metadata.get("logicalType").map(|s| s.as_str()),
2560 Some("uuid"),
2561 "logicalType metadata should be 'uuid'"
2562 );
2563 }
2564
2565 #[test]
2566 fn test_duration_logical_type() {
2567 let mut codec = Codec::Fixed(12);
2568
2569 if let c @ Codec::Fixed(12) = &mut codec {
2570 *c = Codec::Interval;
2571 }
2572
2573 assert!(matches!(codec, Codec::Interval));
2574 }
2575
2576 #[test]
2577 fn test_decimal_logical_type_not_implemented() {
2578 let codec = Codec::Fixed(16);
2579
2580 let process_decimal = || -> Result<(), ArrowError> {
2581 if let Codec::Fixed(_) = codec {
2582 return Err(ArrowError::NotYetImplemented(
2583 "Decimals are not currently supported".to_string(),
2584 ));
2585 }
2586 Ok(())
2587 };
2588
2589 let result = process_decimal();
2590
2591 assert!(result.is_err());
2592 if let Err(ArrowError::NotYetImplemented(msg)) = result {
2593 assert!(msg.contains("Decimals are not currently supported"));
2594 } else {
2595 panic!("Expected NotYetImplemented error");
2596 }
2597 }
2598 #[test]
2599 fn test_unknown_logical_type_added_to_metadata() {
2600 let schema = create_schema_with_logical_type(PrimitiveType::Int, "custom-type");
2601
2602 let mut maker = Maker::new(false, false, Tz::default());
2603 let result = maker.make_data_type(&schema, None, None).unwrap();
2604
2605 assert_eq!(
2606 result.metadata.get("logicalType"),
2607 Some(&"custom-type".to_string())
2608 );
2609 }
2610
2611 #[test]
2612 fn test_string_with_utf8view_enabled() {
2613 let schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
2614
2615 let mut maker = Maker::new(true, false, Tz::default());
2616 let result = maker.make_data_type(&schema, None, None).unwrap();
2617
2618 assert!(matches!(result.codec, Codec::Utf8View));
2619 }
2620
2621 #[test]
2622 fn test_string_without_utf8view_enabled() {
2623 let schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
2624
2625 let mut maker = Maker::new(false, false, Tz::default());
2626 let result = maker.make_data_type(&schema, None, None).unwrap();
2627
2628 assert!(matches!(result.codec, Codec::Utf8));
2629 }
2630
2631 #[test]
2632 fn test_record_with_string_and_utf8view_enabled() {
2633 let field_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
2634
2635 let avro_field = crate::schema::Field {
2636 name: "string_field",
2637 r#type: field_schema,
2638 default: None,
2639 doc: None,
2640 aliases: vec![],
2641 };
2642
2643 let record = Record {
2644 name: "test_record",
2645 namespace: None,
2646 aliases: vec![],
2647 doc: None,
2648 fields: vec![avro_field],
2649 attributes: Attributes::default(),
2650 };
2651
2652 let schema = Schema::Complex(ComplexType::Record(record));
2653
2654 let mut maker = Maker::new(true, false, Tz::default());
2655 let result = maker.make_data_type(&schema, None, None).unwrap();
2656
2657 if let Codec::Struct(fields) = &result.codec {
2658 let first_field_codec = &fields[0].data_type().codec;
2659 assert!(matches!(first_field_codec, Codec::Utf8View));
2660 } else {
2661 panic!("Expected Struct codec");
2662 }
2663 }
2664
2665 #[test]
2666 fn test_union_with_strict_mode() {
2667 let schema = Schema::Union(vec![
2668 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2669 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2670 ]);
2671
2672 let mut maker = Maker::new(false, true, Tz::default());
2673 let result = maker.make_data_type(&schema, None, None);
2674
2675 assert!(result.is_err());
2676 match result {
2677 Err(ArrowError::SchemaError(msg)) => {
2678 assert!(msg.contains(
2679 "Found Avro union of the form ['T','null'], which is disallowed in strict_mode"
2680 ));
2681 }
2682 _ => panic!("Expected SchemaError"),
2683 }
2684 }
2685
2686 #[test]
2687 fn test_resolve_int_to_float_promotion() {
2688 let result = resolve_promotion(PrimitiveType::Int, PrimitiveType::Float);
2689 assert!(matches!(result.codec, Codec::Float32));
2690 assert_eq!(
2691 result.resolution,
2692 Some(ResolutionInfo::Promotion(Promotion::IntToFloat))
2693 );
2694 }
2695
2696 #[test]
2697 fn test_resolve_int_to_double_promotion() {
2698 let result = resolve_promotion(PrimitiveType::Int, PrimitiveType::Double);
2699 assert!(matches!(result.codec, Codec::Float64));
2700 assert_eq!(
2701 result.resolution,
2702 Some(ResolutionInfo::Promotion(Promotion::IntToDouble))
2703 );
2704 }
2705
2706 #[test]
2707 fn test_resolve_long_to_float_promotion() {
2708 let result = resolve_promotion(PrimitiveType::Long, PrimitiveType::Float);
2709 assert!(matches!(result.codec, Codec::Float32));
2710 assert_eq!(
2711 result.resolution,
2712 Some(ResolutionInfo::Promotion(Promotion::LongToFloat))
2713 );
2714 }
2715
2716 #[test]
2717 fn test_resolve_long_to_double_promotion() {
2718 let result = resolve_promotion(PrimitiveType::Long, PrimitiveType::Double);
2719 assert!(matches!(result.codec, Codec::Float64));
2720 assert_eq!(
2721 result.resolution,
2722 Some(ResolutionInfo::Promotion(Promotion::LongToDouble))
2723 );
2724 }
2725
2726 #[test]
2727 fn test_resolve_float_to_double_promotion() {
2728 let result = resolve_promotion(PrimitiveType::Float, PrimitiveType::Double);
2729 assert!(matches!(result.codec, Codec::Float64));
2730 assert_eq!(
2731 result.resolution,
2732 Some(ResolutionInfo::Promotion(Promotion::FloatToDouble))
2733 );
2734 }
2735
2736 #[test]
2737 fn test_resolve_string_to_bytes_promotion() {
2738 let result = resolve_promotion(PrimitiveType::String, PrimitiveType::Bytes);
2739 assert!(matches!(result.codec, Codec::Binary));
2740 assert_eq!(
2741 result.resolution,
2742 Some(ResolutionInfo::Promotion(Promotion::StringToBytes))
2743 );
2744 }
2745
2746 #[test]
2747 fn test_resolve_bytes_to_string_promotion() {
2748 let result = resolve_promotion(PrimitiveType::Bytes, PrimitiveType::String);
2749 assert!(matches!(result.codec, Codec::Utf8));
2750 assert_eq!(
2751 result.resolution,
2752 Some(ResolutionInfo::Promotion(Promotion::BytesToString))
2753 );
2754 }
2755
2756 #[test]
2757 fn test_resolve_illegal_promotion_double_to_float_errors() {
2758 let writer_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Double));
2759 let reader_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Float));
2760 let mut maker = Maker::new(false, false, Tz::default());
2761 let result = maker.make_data_type(&writer_schema, Some(&reader_schema), None);
2762 assert!(result.is_err());
2763 match result {
2764 Err(ArrowError::ParseError(msg)) => {
2765 assert!(msg.contains("Illegal promotion"));
2766 }
2767 _ => panic!("Expected ParseError for illegal promotion Double -> Float"),
2768 }
2769 }
2770
2771 #[test]
2772 fn test_promotion_within_nullable_union_keeps_writer_null_ordering() {
2773 let writer = Schema::Union(vec![
2774 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2775 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2776 ]);
2777 let reader = Schema::Union(vec![
2778 Schema::TypeName(TypeName::Primitive(PrimitiveType::Double)),
2779 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2780 ]);
2781 let mut maker = Maker::new(false, false, Tz::default());
2782 let result = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2783 assert!(matches!(result.codec, Codec::Float64));
2784 assert_eq!(
2785 result.resolution,
2786 Some(ResolutionInfo::Union(ResolvedUnion {
2787 writer_to_reader: [
2788 None,
2789 Some((0, ResolutionInfo::Promotion(Promotion::IntToDouble)))
2790 ]
2791 .into(),
2792 writer_is_union: true,
2793 reader_is_union: true,
2794 }))
2795 );
2796 assert_eq!(result.nullability, Some(Nullability::NullFirst));
2797 }
2798
2799 #[test]
2800 fn test_resolve_writer_union_to_reader_non_union_partial_coverage() {
2801 let writer = mk_union(vec![
2802 mk_primitive(PrimitiveType::String),
2803 mk_primitive(PrimitiveType::Long),
2804 ]);
2805 let reader = mk_primitive(PrimitiveType::Bytes);
2806 let mut maker = Maker::new(false, false, Tz::default());
2807 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2808 assert!(matches!(dt.codec(), Codec::Binary));
2809 let resolved = match dt.resolution {
2810 Some(ResolutionInfo::Union(u)) => u,
2811 other => panic!("expected union resolution info, got {other:?}"),
2812 };
2813 assert!(resolved.writer_is_union && !resolved.reader_is_union);
2814 assert_eq!(
2815 resolved.writer_to_reader.as_ref(),
2816 &[
2817 Some((0, ResolutionInfo::Promotion(Promotion::StringToBytes))),
2818 None
2819 ]
2820 );
2821 }
2822
2823 #[test]
2824 fn test_resolve_writer_non_union_to_reader_union_prefers_direct_over_promotion() {
2825 let writer = mk_primitive(PrimitiveType::Long);
2826 let reader = mk_union(vec![
2827 mk_primitive(PrimitiveType::Long),
2828 mk_primitive(PrimitiveType::Double),
2829 ]);
2830 let mut maker = Maker::new(false, false, Tz::default());
2831 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2832 let resolved = match dt.resolution {
2833 Some(ResolutionInfo::Union(u)) => u,
2834 other => panic!("expected union resolution info, got {other:?}"),
2835 };
2836 assert!(!resolved.writer_is_union && resolved.reader_is_union);
2837 assert_eq!(
2838 resolved.writer_to_reader.as_ref(),
2839 &[Some((0, ResolutionInfo::Promotion(Promotion::Direct)))]
2840 );
2841 }
2842
2843 #[test]
2844 fn test_resolve_writer_non_union_to_reader_union_uses_promotion_when_needed() {
2845 let writer = mk_primitive(PrimitiveType::Int);
2846 let reader = mk_union(vec![
2847 mk_primitive(PrimitiveType::Null),
2848 mk_primitive(PrimitiveType::Long),
2849 mk_primitive(PrimitiveType::String),
2850 ]);
2851 let mut maker = Maker::new(false, false, Tz::default());
2852 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
2853 let resolved = match dt.resolution {
2854 Some(ResolutionInfo::Union(u)) => u,
2855 other => panic!("expected union resolution info, got {other:?}"),
2856 };
2857 assert_eq!(
2858 resolved.writer_to_reader.as_ref(),
2859 &[Some((1, ResolutionInfo::Promotion(Promotion::IntToLong)))]
2860 );
2861 }
2862
2863 #[test]
2864 fn test_resolve_writer_non_union_to_reader_union_preserves_inner_record_defaults() {
2865 let writer = Schema::Complex(ComplexType::Record(Record {
2869 name: "Inner",
2870 namespace: None,
2871 doc: None,
2872 aliases: vec![],
2873 fields: vec![AvroFieldSchema {
2874 name: "a",
2875 doc: None,
2876 r#type: mk_primitive(PrimitiveType::Int),
2877 default: None,
2878 aliases: vec![],
2879 }],
2880 attributes: Attributes::default(),
2881 }));
2882 let reader = mk_union(vec![
2883 Schema::Complex(ComplexType::Record(Record {
2884 name: "Inner",
2885 namespace: None,
2886 doc: None,
2887 aliases: vec![],
2888 fields: vec![
2889 AvroFieldSchema {
2890 name: "a",
2891 doc: None,
2892 r#type: mk_primitive(PrimitiveType::Int),
2893 default: None,
2894 aliases: vec![],
2895 },
2896 AvroFieldSchema {
2897 name: "b",
2898 doc: None,
2899 r#type: mk_primitive(PrimitiveType::Int),
2900 default: Some(Value::Number(serde_json::Number::from(42))),
2901 aliases: vec![],
2902 },
2903 ],
2904 attributes: Attributes::default(),
2905 })),
2906 mk_primitive(PrimitiveType::String),
2907 ]);
2908 let mut maker = Maker::new(false, false, Default::default());
2909 let dt = maker
2910 .make_data_type(&writer, Some(&reader), None)
2911 .expect("resolution should succeed");
2912 let resolved = match dt.resolution.as_ref() {
2914 Some(ResolutionInfo::Union(u)) => u,
2915 other => panic!("expected union resolution info, got {other:?}"),
2916 };
2917 assert!(!resolved.writer_is_union && resolved.reader_is_union);
2918 assert_eq!(
2919 resolved.writer_to_reader.len(),
2920 1,
2921 "expected the non-union record to resolve to a union variant"
2922 );
2923 let resolution = match resolved.writer_to_reader.first().unwrap() {
2924 Some((0, resolution)) => resolution,
2925 other => panic!("unexpected writer-to-reader table value {other:?}"),
2926 };
2927 match resolution {
2928 ResolutionInfo::Record(ResolvedRecord {
2929 writer_fields,
2930 default_fields,
2931 }) => {
2932 assert_eq!(writer_fields.len(), 1);
2933 assert!(matches!(writer_fields[0], ResolvedField::ToReader(0, _)));
2934 assert_eq!(default_fields.len(), 1);
2935 assert_eq!(default_fields[0], 1);
2936 }
2937 other => panic!("unexpected resolution {other:?}"),
2938 }
2939 let children = match dt.codec() {
2941 Codec::Union(children, _, _) => children,
2942 other => panic!("expected union codec, got {other:?}"),
2943 };
2944 let inner_fields = match children[0].codec() {
2945 Codec::Struct(f) => f,
2946 other => panic!("expected struct codec for Inner, got {other:?}"),
2947 };
2948 assert_eq!(inner_fields.len(), 2);
2949 assert_eq!(inner_fields[1].name(), "b");
2950 assert_eq!(
2951 inner_fields[1].data_type().resolution,
2952 Some(ResolutionInfo::DefaultValue(AvroLiteral::Int(42))),
2953 "field b should have DefaultValue(Int(42)) from schema resolution"
2954 );
2955 }
2956
2957 #[test]
2958 fn test_resolve_writer_union_to_reader_union_preserves_inner_record_defaults() {
2959 let writer = mk_union(vec![
2963 mk_primitive(PrimitiveType::String),
2964 Schema::Complex(ComplexType::Record(Record {
2965 name: "Inner",
2966 namespace: None,
2967 doc: None,
2968 aliases: vec![],
2969 fields: vec![AvroFieldSchema {
2970 name: "a",
2971 doc: None,
2972 r#type: mk_primitive(PrimitiveType::Int),
2973 default: None,
2974 aliases: vec![],
2975 }],
2976 attributes: Attributes::default(),
2977 })),
2978 ]);
2979 let reader = mk_union(vec![
2980 Schema::Complex(ComplexType::Record(Record {
2981 name: "Inner",
2982 namespace: None,
2983 doc: None,
2984 aliases: vec![],
2985 fields: vec![
2986 AvroFieldSchema {
2987 name: "a",
2988 doc: None,
2989 r#type: mk_primitive(PrimitiveType::Int),
2990 default: None,
2991 aliases: vec![],
2992 },
2993 AvroFieldSchema {
2994 name: "b",
2995 doc: None,
2996 r#type: mk_primitive(PrimitiveType::Int),
2997 default: Some(Value::Number(serde_json::Number::from(42))),
2998 aliases: vec![],
2999 },
3000 ],
3001 attributes: Attributes::default(),
3002 })),
3003 mk_primitive(PrimitiveType::String),
3004 ]);
3005 let mut maker = Maker::new(false, false, Default::default());
3006 let dt = maker
3007 .make_data_type(&writer, Some(&reader), None)
3008 .expect("resolution should succeed");
3009 let resolved = match dt.resolution.as_ref() {
3011 Some(ResolutionInfo::Union(u)) => u,
3012 other => panic!("expected union resolution info, got {other:?}"),
3013 };
3014 assert!(resolved.writer_is_union && resolved.reader_is_union);
3015 assert_eq!(resolved.writer_to_reader.len(), 2);
3016 let resolution = match resolved.writer_to_reader[1].as_ref() {
3017 Some((0, resolution)) => resolution,
3018 other => panic!("unexpected writer-to-reader table value {other:?}"),
3019 };
3020 match resolution {
3021 ResolutionInfo::Record(ResolvedRecord {
3022 writer_fields,
3023 default_fields,
3024 }) => {
3025 assert_eq!(writer_fields.len(), 1);
3026 assert!(matches!(writer_fields[0], ResolvedField::ToReader(0, _)));
3027 assert_eq!(default_fields.len(), 1);
3028 assert_eq!(default_fields[0], 1);
3029 }
3030 other => panic!("unexpected resolution {other:?}"),
3031 }
3032 let children = match dt.codec() {
3034 Codec::Union(children, _, _) => children,
3035 other => panic!("expected union codec, got {other:?}"),
3036 };
3037 let inner_fields = match children[0].codec() {
3038 Codec::Struct(f) => f,
3039 other => panic!("expected struct codec for Inner, got {other:?}"),
3040 };
3041 assert_eq!(inner_fields.len(), 2);
3042 assert_eq!(inner_fields[1].name(), "b");
3043 assert_eq!(
3044 inner_fields[1].data_type().resolution,
3045 Some(ResolutionInfo::DefaultValue(AvroLiteral::Int(42))),
3046 "field b should have DefaultValue(Int(42)) from schema resolution"
3047 );
3048 }
3049
3050 #[test]
3051 fn test_resolve_both_nullable_unions_direct_match() {
3052 let writer = mk_union(vec![
3053 mk_primitive(PrimitiveType::Null),
3054 mk_primitive(PrimitiveType::String),
3055 ]);
3056 let reader = mk_union(vec![
3057 mk_primitive(PrimitiveType::String),
3058 mk_primitive(PrimitiveType::Null),
3059 ]);
3060 let mut maker = Maker::new(false, false, Tz::default());
3061 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
3062 assert!(matches!(dt.codec(), Codec::Utf8));
3063 assert_eq!(dt.nullability, Some(Nullability::NullFirst));
3064 assert_eq!(
3065 dt.resolution,
3066 Some(ResolutionInfo::Union(ResolvedUnion {
3067 writer_to_reader: [
3068 None,
3069 Some((0, ResolutionInfo::Promotion(Promotion::Direct)))
3070 ]
3071 .into(),
3072 writer_is_union: true,
3073 reader_is_union: true
3074 }))
3075 );
3076 }
3077
3078 #[test]
3079 fn test_resolve_both_nullable_unions_with_promotion() {
3080 let writer = mk_union(vec![
3081 mk_primitive(PrimitiveType::Null),
3082 mk_primitive(PrimitiveType::Int),
3083 ]);
3084 let reader = mk_union(vec![
3085 mk_primitive(PrimitiveType::Double),
3086 mk_primitive(PrimitiveType::Null),
3087 ]);
3088 let mut maker = Maker::new(false, false, Tz::default());
3089 let dt = maker.make_data_type(&writer, Some(&reader), None).unwrap();
3090 assert!(matches!(dt.codec(), Codec::Float64));
3091 assert_eq!(dt.nullability, Some(Nullability::NullFirst));
3092 assert_eq!(
3093 dt.resolution,
3094 Some(ResolutionInfo::Union(ResolvedUnion {
3095 writer_to_reader: [
3096 None,
3097 Some((0, ResolutionInfo::Promotion(Promotion::IntToDouble)))
3098 ]
3099 .into(),
3100 writer_is_union: true,
3101 reader_is_union: true
3102 }))
3103 );
3104 }
3105
3106 #[test]
3107 fn test_resolve_type_promotion() {
3108 let writer_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3109 let reader_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::Long));
3110 let mut maker = Maker::new(false, false, Tz::default());
3111 let result = maker
3112 .make_data_type(&writer_schema, Some(&reader_schema), None)
3113 .unwrap();
3114 assert!(matches!(result.codec, Codec::Int64));
3115 assert_eq!(
3116 result.resolution,
3117 Some(ResolutionInfo::Promotion(Promotion::IntToLong))
3118 );
3119 }
3120
3121 #[test]
3122 fn test_nested_record_type_reuse_without_namespace() {
3123 let schema_str = r#"
3124 {
3125 "type": "record",
3126 "name": "Record",
3127 "fields": [
3128 {
3129 "name": "nested",
3130 "type": {
3131 "type": "record",
3132 "name": "Nested",
3133 "fields": [
3134 { "name": "nested_int", "type": "int" }
3135 ]
3136 }
3137 },
3138 { "name": "nestedRecord", "type": "Nested" },
3139 { "name": "nestedArray", "type": { "type": "array", "items": "Nested" } },
3140 { "name": "nestedMap", "type": { "type": "map", "values": "Nested" } }
3141 ]
3142 }
3143 "#;
3144
3145 let schema: Schema = serde_json::from_str(schema_str).unwrap();
3146
3147 let mut maker = Maker::new(false, false, Tz::default());
3148 let avro_data_type = maker.make_data_type(&schema, None, None).unwrap();
3149
3150 if let Codec::Struct(fields) = avro_data_type.codec() {
3151 assert_eq!(fields.len(), 4);
3152
3153 assert_eq!(fields[0].name(), "nested");
3155 let nested_data_type = fields[0].data_type();
3156 if let Codec::Struct(nested_fields) = nested_data_type.codec() {
3157 assert_eq!(nested_fields.len(), 1);
3158 assert_eq!(nested_fields[0].name(), "nested_int");
3159 assert!(matches!(nested_fields[0].data_type().codec(), Codec::Int32));
3160 } else {
3161 panic!(
3162 "'nested' field is not a struct but {:?}",
3163 nested_data_type.codec()
3164 );
3165 }
3166
3167 assert_eq!(fields[1].name(), "nestedRecord");
3169 let nested_record_data_type = fields[1].data_type();
3170 assert_eq!(
3171 nested_record_data_type.codec().data_type(),
3172 nested_data_type.codec().data_type()
3173 );
3174
3175 assert_eq!(fields[2].name(), "nestedArray");
3177 if let Codec::List(item_type) = fields[2].data_type().codec() {
3178 assert_eq!(
3179 item_type.codec().data_type(),
3180 nested_data_type.codec().data_type()
3181 );
3182 } else {
3183 panic!("'nestedArray' field is not a list");
3184 }
3185
3186 assert_eq!(fields[3].name(), "nestedMap");
3188 if let Codec::Map(value_type) = fields[3].data_type().codec() {
3189 assert_eq!(
3190 value_type.codec().data_type(),
3191 nested_data_type.codec().data_type()
3192 );
3193 } else {
3194 panic!("'nestedMap' field is not a map");
3195 }
3196 } else {
3197 panic!("Top-level schema is not a struct");
3198 }
3199 }
3200
3201 #[test]
3202 fn test_nested_enum_type_reuse_with_namespace() {
3203 let schema_str = r#"
3204 {
3205 "type": "record",
3206 "name": "Record",
3207 "namespace": "record_ns",
3208 "fields": [
3209 {
3210 "name": "status",
3211 "type": {
3212 "type": "enum",
3213 "name": "Status",
3214 "namespace": "enum_ns",
3215 "symbols": ["ACTIVE", "INACTIVE", "PENDING"]
3216 }
3217 },
3218 { "name": "backupStatus", "type": "enum_ns.Status" },
3219 { "name": "statusHistory", "type": { "type": "array", "items": "enum_ns.Status" } },
3220 { "name": "statusMap", "type": { "type": "map", "values": "enum_ns.Status" } }
3221 ]
3222 }
3223 "#;
3224
3225 let schema: Schema = serde_json::from_str(schema_str).unwrap();
3226
3227 let mut maker = Maker::new(false, false, Tz::default());
3228 let avro_data_type = maker.make_data_type(&schema, None, None).unwrap();
3229
3230 if let Codec::Struct(fields) = avro_data_type.codec() {
3231 assert_eq!(fields.len(), 4);
3232
3233 assert_eq!(fields[0].name(), "status");
3235 let status_data_type = fields[0].data_type();
3236 if let Codec::Enum(symbols) = status_data_type.codec() {
3237 assert_eq!(symbols.as_ref(), &["ACTIVE", "INACTIVE", "PENDING"]);
3238 } else {
3239 panic!(
3240 "'status' field is not an enum but {:?}",
3241 status_data_type.codec()
3242 );
3243 }
3244
3245 assert_eq!(fields[1].name(), "backupStatus");
3247 let backup_status_data_type = fields[1].data_type();
3248 assert_eq!(
3249 backup_status_data_type.codec().data_type(),
3250 status_data_type.codec().data_type()
3251 );
3252
3253 assert_eq!(fields[2].name(), "statusHistory");
3255 if let Codec::List(item_type) = fields[2].data_type().codec() {
3256 assert_eq!(
3257 item_type.codec().data_type(),
3258 status_data_type.codec().data_type()
3259 );
3260 } else {
3261 panic!("'statusHistory' field is not a list");
3262 }
3263
3264 assert_eq!(fields[3].name(), "statusMap");
3266 if let Codec::Map(value_type) = fields[3].data_type().codec() {
3267 assert_eq!(
3268 value_type.codec().data_type(),
3269 status_data_type.codec().data_type()
3270 );
3271 } else {
3272 panic!("'statusMap' field is not a map");
3273 }
3274 } else {
3275 panic!("Top-level schema is not a struct");
3276 }
3277 }
3278
3279 #[test]
3280 fn test_resolve_from_writer_and_reader_defaults_root_name_for_non_record_reader() {
3281 let writer_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
3282 let reader_schema = Schema::TypeName(TypeName::Primitive(PrimitiveType::String));
3283 let mut maker = Maker::new(false, false, Tz::default());
3284 let data_type = maker
3285 .make_data_type(&writer_schema, Some(&reader_schema), None)
3286 .expect("resolution should succeed");
3287 let field = AvroField {
3288 name: AVRO_ROOT_RECORD_DEFAULT_NAME.to_string(),
3289 data_type,
3290 };
3291 assert_eq!(field.name(), AVRO_ROOT_RECORD_DEFAULT_NAME);
3292 assert!(matches!(field.data_type().codec(), Codec::Utf8));
3293 }
3294
3295 fn json_string(s: &str) -> Value {
3296 Value::String(s.to_string())
3297 }
3298
3299 fn assert_default_stored(dt: &AvroDataType, default_json: &Value) {
3300 let stored = dt
3301 .metadata
3302 .get(AVRO_FIELD_DEFAULT_METADATA_KEY)
3303 .cloned()
3304 .unwrap_or_default();
3305 let expected = serde_json::to_string(default_json).unwrap();
3306 assert_eq!(stored, expected, "stored default metadata should match");
3307 }
3308
3309 #[test]
3310 fn test_validate_and_store_default_null_and_nullability_rules() {
3311 let mut dt_null = AvroDataType::new(Codec::Null, HashMap::new(), None);
3312 let lit = dt_null.parse_and_store_default(&Value::Null).unwrap();
3313 assert_eq!(lit, AvroLiteral::Null);
3314 assert_default_stored(&dt_null, &Value::Null);
3315 let mut dt_int = AvroDataType::new(Codec::Int32, HashMap::new(), None);
3316 let err = dt_int.parse_and_store_default(&Value::Null).unwrap_err();
3317 assert!(
3318 err.to_string()
3319 .contains("JSON null default is only valid for `null` type"),
3320 "unexpected error: {err}"
3321 );
3322 let mut dt_int_nf =
3323 AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullFirst));
3324 let lit2 = dt_int_nf.parse_and_store_default(&Value::Null).unwrap();
3325 assert_eq!(lit2, AvroLiteral::Null);
3326 assert_default_stored(&dt_int_nf, &Value::Null);
3327 let mut dt_int_ns =
3328 AvroDataType::new(Codec::Int32, HashMap::new(), Some(Nullability::NullSecond));
3329 let err2 = dt_int_ns.parse_and_store_default(&Value::Null).unwrap_err();
3330 assert!(
3331 err2.to_string()
3332 .contains("JSON null default is only valid for `null` type"),
3333 "unexpected error: {err2}"
3334 );
3335 }
3336
3337 #[test]
3338 fn test_validate_and_store_default_primitives_and_temporal() {
3339 let mut dt_bool = AvroDataType::new(Codec::Boolean, HashMap::new(), None);
3340 let lit = dt_bool.parse_and_store_default(&Value::Bool(true)).unwrap();
3341 assert_eq!(lit, AvroLiteral::Boolean(true));
3342 assert_default_stored(&dt_bool, &Value::Bool(true));
3343 let mut dt_i32 = AvroDataType::new(Codec::Int32, HashMap::new(), None);
3344 let lit = dt_i32
3345 .parse_and_store_default(&serde_json::json!(123))
3346 .unwrap();
3347 assert_eq!(lit, AvroLiteral::Int(123));
3348 assert_default_stored(&dt_i32, &serde_json::json!(123));
3349 let err = dt_i32
3350 .parse_and_store_default(&serde_json::json!(i64::from(i32::MAX) + 1))
3351 .unwrap_err();
3352 assert!(format!("{err}").contains("out of i32 range"));
3353 let mut dt_i64 = AvroDataType::new(Codec::Int64, HashMap::new(), None);
3354 let lit = dt_i64
3355 .parse_and_store_default(&serde_json::json!(1234567890))
3356 .unwrap();
3357 assert_eq!(lit, AvroLiteral::Long(1234567890));
3358 assert_default_stored(&dt_i64, &serde_json::json!(1234567890));
3359 let mut dt_f32 = AvroDataType::new(Codec::Float32, HashMap::new(), None);
3360 let lit = dt_f32
3361 .parse_and_store_default(&serde_json::json!(1.25))
3362 .unwrap();
3363 assert_eq!(lit, AvroLiteral::Float(1.25));
3364 assert_default_stored(&dt_f32, &serde_json::json!(1.25));
3365 let err = dt_f32
3366 .parse_and_store_default(&serde_json::json!(1e39))
3367 .unwrap_err();
3368 assert!(format!("{err}").contains("out of f32 range"));
3369 let mut dt_f64 = AvroDataType::new(Codec::Float64, HashMap::new(), None);
3370 let lit = dt_f64
3371 .parse_and_store_default(&serde_json::json!(std::f64::consts::PI))
3372 .unwrap();
3373 assert_eq!(lit, AvroLiteral::Double(std::f64::consts::PI));
3374 assert_default_stored(&dt_f64, &serde_json::json!(std::f64::consts::PI));
3375 let mut dt_str = AvroDataType::new(Codec::Utf8, HashMap::new(), None);
3376 let l = dt_str
3377 .parse_and_store_default(&json_string("hello"))
3378 .unwrap();
3379 assert_eq!(l, AvroLiteral::String("hello".into()));
3380 assert_default_stored(&dt_str, &json_string("hello"));
3381 let mut dt_strv = AvroDataType::new(Codec::Utf8View, HashMap::new(), None);
3382 let l = dt_strv
3383 .parse_and_store_default(&json_string("view"))
3384 .unwrap();
3385 assert_eq!(l, AvroLiteral::String("view".into()));
3386 assert_default_stored(&dt_strv, &json_string("view"));
3387 let mut dt_uuid = AvroDataType::new(Codec::Uuid, HashMap::new(), None);
3388 let l = dt_uuid
3389 .parse_and_store_default(&json_string("00000000-0000-0000-0000-000000000000"))
3390 .unwrap();
3391 assert_eq!(
3392 l,
3393 AvroLiteral::String("00000000-0000-0000-0000-000000000000".into())
3394 );
3395 let mut dt_bin = AvroDataType::new(Codec::Binary, HashMap::new(), None);
3396 let l = dt_bin.parse_and_store_default(&json_string("ABC")).unwrap();
3397 assert_eq!(l, AvroLiteral::Bytes(vec![65, 66, 67]));
3398 let err = dt_bin
3399 .parse_and_store_default(&json_string("€")) .unwrap_err();
3401 assert!(format!("{err}").contains("Invalid codepoint"));
3402 let mut dt_date = AvroDataType::new(Codec::Date32, HashMap::new(), None);
3403 let ld = dt_date
3404 .parse_and_store_default(&serde_json::json!(1))
3405 .unwrap();
3406 assert_eq!(ld, AvroLiteral::Int(1));
3407 let mut dt_tmill = AvroDataType::new(Codec::TimeMillis, HashMap::new(), None);
3408 let lt = dt_tmill
3409 .parse_and_store_default(&serde_json::json!(86_400_000))
3410 .unwrap();
3411 assert_eq!(lt, AvroLiteral::Int(86_400_000));
3412 let mut dt_tmicros = AvroDataType::new(Codec::TimeMicros, HashMap::new(), None);
3413 let ltm = dt_tmicros
3414 .parse_and_store_default(&serde_json::json!(1_000_000))
3415 .unwrap();
3416 assert_eq!(ltm, AvroLiteral::Long(1_000_000));
3417 let mut dt_ts_milli = AvroDataType::new(Codec::TimestampMillis(None), HashMap::new(), None);
3418 let l1 = dt_ts_milli
3419 .parse_and_store_default(&serde_json::json!(123))
3420 .unwrap();
3421 assert_eq!(l1, AvroLiteral::Long(123));
3422 let mut dt_ts_micro = AvroDataType::new(Codec::TimestampMicros(None), HashMap::new(), None);
3423 let l2 = dt_ts_micro
3424 .parse_and_store_default(&serde_json::json!(456))
3425 .unwrap();
3426 assert_eq!(l2, AvroLiteral::Long(456));
3427 }
3428
3429 #[cfg(feature = "avro_custom_types")]
3430 #[test]
3431 fn test_validate_and_store_default_custom_integer_ranges() {
3432 let mut dt_i8 = AvroDataType::new(Codec::Int8, HashMap::new(), None);
3433 let lit_i8 = dt_i8
3434 .parse_and_store_default(&serde_json::json!(i8::MAX))
3435 .unwrap();
3436 assert_eq!(lit_i8, AvroLiteral::Int(i8::MAX as i32));
3437 let err_i8_high = dt_i8
3438 .parse_and_store_default(&serde_json::json!(i8::MAX as i64 + 1))
3439 .unwrap_err();
3440 assert!(err_i8_high.to_string().contains("out of i8 range"));
3441 let err_i8_low = dt_i8
3442 .parse_and_store_default(&serde_json::json!(i8::MIN as i64 - 1))
3443 .unwrap_err();
3444 assert!(err_i8_low.to_string().contains("out of i8 range"));
3445
3446 let mut dt_i16 = AvroDataType::new(Codec::Int16, HashMap::new(), None);
3447 let lit_i16 = dt_i16
3448 .parse_and_store_default(&serde_json::json!(i16::MIN))
3449 .unwrap();
3450 assert_eq!(lit_i16, AvroLiteral::Int(i16::MIN as i32));
3451 let err_i16_high = dt_i16
3452 .parse_and_store_default(&serde_json::json!(i16::MAX as i64 + 1))
3453 .unwrap_err();
3454 assert!(err_i16_high.to_string().contains("out of i16 range"));
3455 let err_i16_low = dt_i16
3456 .parse_and_store_default(&serde_json::json!(i16::MIN as i64 - 1))
3457 .unwrap_err();
3458 assert!(err_i16_low.to_string().contains("out of i16 range"));
3459
3460 let mut dt_u8 = AvroDataType::new(Codec::UInt8, HashMap::new(), None);
3461 let lit_u8 = dt_u8
3462 .parse_and_store_default(&serde_json::json!(u8::MAX))
3463 .unwrap();
3464 assert_eq!(lit_u8, AvroLiteral::Int(u8::MAX as i32));
3465 let err_u8_neg = dt_u8
3466 .parse_and_store_default(&serde_json::json!(-1))
3467 .unwrap_err();
3468 assert!(err_u8_neg.to_string().contains("out of u8 range"));
3469 let err_u8_high = dt_u8
3470 .parse_and_store_default(&serde_json::json!(u8::MAX as i64 + 1))
3471 .unwrap_err();
3472 assert!(err_u8_high.to_string().contains("out of u8 range"));
3473
3474 let mut dt_u16 = AvroDataType::new(Codec::UInt16, HashMap::new(), None);
3475 let lit_u16 = dt_u16
3476 .parse_and_store_default(&serde_json::json!(u16::MAX))
3477 .unwrap();
3478 assert_eq!(lit_u16, AvroLiteral::Int(u16::MAX as i32));
3479 let err_u16_neg = dt_u16
3480 .parse_and_store_default(&serde_json::json!(-1))
3481 .unwrap_err();
3482 assert!(err_u16_neg.to_string().contains("out of u16 range"));
3483 let err_u16_high = dt_u16
3484 .parse_and_store_default(&serde_json::json!(u16::MAX as i64 + 1))
3485 .unwrap_err();
3486 assert!(err_u16_high.to_string().contains("out of u16 range"));
3487
3488 let mut dt_u32 = AvroDataType::new(Codec::UInt32, HashMap::new(), None);
3489 let lit_u32 = dt_u32
3490 .parse_and_store_default(&serde_json::json!(u32::MAX as i64))
3491 .unwrap();
3492 assert_eq!(lit_u32, AvroLiteral::Long(u32::MAX as i64));
3493 let err_u32_neg = dt_u32
3494 .parse_and_store_default(&serde_json::json!(-1))
3495 .unwrap_err();
3496 assert!(err_u32_neg.to_string().contains("out of u32 range"));
3497 let err_u32_high = dt_u32
3498 .parse_and_store_default(&serde_json::json!(u32::MAX as i64 + 1))
3499 .unwrap_err();
3500 assert!(err_u32_high.to_string().contains("out of u32 range"));
3501 }
3502
3503 #[test]
3504 fn test_validate_and_store_default_fixed_decimal_interval() {
3505 let mut dt_fixed = AvroDataType::new(Codec::Fixed(4), HashMap::new(), None);
3506 let l = dt_fixed
3507 .parse_and_store_default(&json_string("WXYZ"))
3508 .unwrap();
3509 assert_eq!(l, AvroLiteral::Bytes(vec![87, 88, 89, 90]));
3510 let err = dt_fixed
3511 .parse_and_store_default(&json_string("TOO LONG"))
3512 .unwrap_err();
3513 assert!(err.to_string().contains("Default length"));
3514 let mut dt_dec_fixed =
3515 AvroDataType::new(Codec::Decimal(10, Some(2), Some(3)), HashMap::new(), None);
3516 let l = dt_dec_fixed
3517 .parse_and_store_default(&json_string("abc"))
3518 .unwrap();
3519 assert_eq!(l, AvroLiteral::Bytes(vec![97, 98, 99]));
3520 let err = dt_dec_fixed
3521 .parse_and_store_default(&json_string("toolong"))
3522 .unwrap_err();
3523 assert!(err.to_string().contains("Default length"));
3524 let mut dt_dec_bytes =
3525 AvroDataType::new(Codec::Decimal(10, Some(2), None), HashMap::new(), None);
3526 let l = dt_dec_bytes
3527 .parse_and_store_default(&json_string("freeform"))
3528 .unwrap();
3529 assert_eq!(
3530 l,
3531 AvroLiteral::Bytes("freeform".bytes().collect::<Vec<_>>())
3532 );
3533 let mut dt_interval = AvroDataType::new(Codec::Interval, HashMap::new(), None);
3534 let l = dt_interval
3535 .parse_and_store_default(&json_string("ABCDEFGHIJKL"))
3536 .unwrap();
3537 assert_eq!(
3538 l,
3539 AvroLiteral::Bytes("ABCDEFGHIJKL".bytes().collect::<Vec<_>>())
3540 );
3541 let err = dt_interval
3542 .parse_and_store_default(&json_string("short"))
3543 .unwrap_err();
3544 assert!(err.to_string().contains("Default length"));
3545 }
3546
3547 #[test]
3548 fn test_validate_and_store_default_enum_list_map_struct() {
3549 let symbols: Arc<[String]> = ["RED".to_string(), "GREEN".to_string(), "BLUE".to_string()]
3550 .into_iter()
3551 .collect();
3552 let mut dt_enum = AvroDataType::new(Codec::Enum(symbols), HashMap::new(), None);
3553 let l = dt_enum
3554 .parse_and_store_default(&json_string("GREEN"))
3555 .unwrap();
3556 assert_eq!(l, AvroLiteral::Enum("GREEN".into()));
3557 let err = dt_enum
3558 .parse_and_store_default(&json_string("YELLOW"))
3559 .unwrap_err();
3560 assert!(err.to_string().contains("Default enum symbol"));
3561 let item = AvroDataType::new(Codec::Int64, HashMap::new(), None);
3562 let mut dt_list = AvroDataType::new(Codec::List(Arc::new(item)), HashMap::new(), None);
3563 let val = serde_json::json!([1, 2, 3]);
3564 let l = dt_list.parse_and_store_default(&val).unwrap();
3565 assert_eq!(
3566 l,
3567 AvroLiteral::Array(vec![
3568 AvroLiteral::Long(1),
3569 AvroLiteral::Long(2),
3570 AvroLiteral::Long(3)
3571 ])
3572 );
3573 let err = dt_list
3574 .parse_and_store_default(&serde_json::json!({"not":"array"}))
3575 .unwrap_err();
3576 assert!(err.to_string().contains("JSON array"));
3577 let val_dt = AvroDataType::new(Codec::Float64, HashMap::new(), None);
3578 let mut dt_map = AvroDataType::new(Codec::Map(Arc::new(val_dt)), HashMap::new(), None);
3579 let mv = serde_json::json!({"x": 1.5, "y": 2.5});
3580 let l = dt_map.parse_and_store_default(&mv).unwrap();
3581 let mut expected = IndexMap::new();
3582 expected.insert("x".into(), AvroLiteral::Double(1.5));
3583 expected.insert("y".into(), AvroLiteral::Double(2.5));
3584 assert_eq!(l, AvroLiteral::Map(expected));
3585 let err = dt_map
3587 .parse_and_store_default(&serde_json::json!(123))
3588 .unwrap_err();
3589 assert!(err.to_string().contains("JSON object"));
3590 let mut field_a = AvroField {
3591 name: "a".into(),
3592 data_type: AvroDataType::new(Codec::Int32, HashMap::new(), None),
3593 };
3594 let field_b = AvroField {
3595 name: "b".into(),
3596 data_type: AvroDataType::new(
3597 Codec::Int64,
3598 HashMap::new(),
3599 Some(Nullability::NullFirst),
3600 ),
3601 };
3602 let mut c_md = HashMap::new();
3603 c_md.insert(AVRO_FIELD_DEFAULT_METADATA_KEY.into(), "\"xyz\"".into());
3604 let field_c = AvroField {
3605 name: "c".into(),
3606 data_type: AvroDataType::new(Codec::Utf8, c_md, None),
3607 };
3608 field_a.data_type.metadata.insert("doc".into(), "na".into());
3609 let struct_fields: Arc<[AvroField]> = Arc::from(vec![field_a, field_b, field_c]);
3610 let mut dt_struct = AvroDataType::new(Codec::Struct(struct_fields), HashMap::new(), None);
3611 let default_obj = serde_json::json!({"a": 7});
3612 let l = dt_struct.parse_and_store_default(&default_obj).unwrap();
3613 let mut expected = IndexMap::new();
3614 expected.insert("a".into(), AvroLiteral::Int(7));
3615 expected.insert("b".into(), AvroLiteral::Null);
3616 expected.insert("c".into(), AvroLiteral::String("xyz".into()));
3617 assert_eq!(l, AvroLiteral::Map(expected));
3618 assert_default_stored(&dt_struct, &default_obj);
3619 let req_field = AvroField {
3620 name: "req".into(),
3621 data_type: AvroDataType::new(Codec::Boolean, HashMap::new(), None),
3622 };
3623 let mut dt_bad = AvroDataType::new(
3624 Codec::Struct(Arc::from(vec![req_field])),
3625 HashMap::new(),
3626 None,
3627 );
3628 let err = dt_bad
3629 .parse_and_store_default(&serde_json::json!({}))
3630 .unwrap_err();
3631 assert!(
3632 err.to_string().contains("missing required subfield 'req'"),
3633 "unexpected error: {err}"
3634 );
3635 let err = dt_struct
3636 .parse_and_store_default(&serde_json::json!(10))
3637 .unwrap_err();
3638 err.to_string().contains("must be a JSON object");
3639 }
3640
3641 #[test]
3642 fn test_resolve_array_promotion_and_reader_metadata() {
3643 let mut w_add: HashMap<&str, Value> = HashMap::new();
3644 w_add.insert("who", json_string("writer"));
3645 let mut r_add: HashMap<&str, Value> = HashMap::new();
3646 r_add.insert("who", json_string("reader"));
3647 let writer_schema = Schema::Complex(ComplexType::Array(Array {
3648 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
3649 attributes: Attributes {
3650 logical_type: None,
3651 additional: w_add,
3652 },
3653 }));
3654 let reader_schema = Schema::Complex(ComplexType::Array(Array {
3655 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Long))),
3656 attributes: Attributes {
3657 logical_type: None,
3658 additional: r_add,
3659 },
3660 }));
3661 let mut maker = Maker::new(false, false, Tz::default());
3662 let dt = maker
3663 .make_data_type(&writer_schema, Some(&reader_schema), None)
3664 .unwrap();
3665 assert_eq!(dt.metadata.get("who"), Some(&"\"reader\"".to_string()));
3666 if let Codec::List(inner) = dt.codec() {
3667 assert!(matches!(inner.codec(), Codec::Int64));
3668 assert_eq!(
3669 inner.resolution,
3670 Some(ResolutionInfo::Promotion(Promotion::IntToLong))
3671 );
3672 } else {
3673 panic!("expected list codec");
3674 }
3675 }
3676
3677 #[test]
3678 fn test_resolve_array_writer_nonunion_items_reader_nullable_items() {
3679 let writer_schema = Schema::Complex(ComplexType::Array(Array {
3680 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
3681 attributes: Attributes::default(),
3682 }));
3683 let reader_schema = Schema::Complex(ComplexType::Array(Array {
3684 items: Box::new(mk_union(vec![
3685 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
3686 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3687 ])),
3688 attributes: Attributes::default(),
3689 }));
3690 let mut maker = Maker::new(false, false, Tz::default());
3691 let dt = maker
3692 .make_data_type(&writer_schema, Some(&reader_schema), None)
3693 .unwrap();
3694 if let Codec::List(inner) = dt.codec() {
3695 assert_eq!(inner.nullability(), Some(Nullability::NullFirst));
3696 assert!(matches!(inner.codec(), Codec::Int32));
3697 match inner.resolution.as_ref() {
3698 Some(ResolutionInfo::Promotion(Promotion::Direct)) => {}
3699 other => panic!("expected Union resolution, got {other:?}"),
3700 }
3701 } else {
3702 panic!("expected List codec");
3703 }
3704 }
3705
3706 #[test]
3707 fn test_resolve_fixed_success_name_and_size_match_and_alias() {
3708 let writer_schema = Schema::Complex(ComplexType::Fixed(Fixed {
3709 name: "MD5",
3710 namespace: None,
3711 aliases: vec!["Hash16"],
3712 size: 16,
3713 attributes: Attributes::default(),
3714 }));
3715 let reader_schema = Schema::Complex(ComplexType::Fixed(Fixed {
3716 name: "Hash16",
3717 namespace: None,
3718 aliases: vec![],
3719 size: 16,
3720 attributes: Attributes::default(),
3721 }));
3722 let mut maker = Maker::new(false, false, Tz::default());
3723 let dt = maker
3724 .make_data_type(&writer_schema, Some(&reader_schema), None)
3725 .unwrap();
3726 assert!(matches!(dt.codec(), Codec::Fixed(16)));
3727 }
3728
3729 #[cfg(feature = "avro_custom_types")]
3730 #[test]
3731 fn test_interval_month_day_nano_custom_logical_type_fixed16() {
3732 let schema = Schema::Complex(ComplexType::Fixed(Fixed {
3733 name: "ArrowIntervalMDN",
3734 namespace: None,
3735 aliases: vec![],
3736 size: 16,
3737 attributes: Attributes {
3738 logical_type: Some("arrow.interval-month-day-nano"),
3739 additional: Default::default(),
3740 },
3741 }));
3742 let mut maker = Maker::new(false, false, Default::default());
3743 let dt = maker.make_data_type(&schema, None, None).unwrap();
3744 assert!(matches!(dt.codec(), Codec::IntervalMonthDayNano));
3745 assert_eq!(
3746 dt.codec.data_type(),
3747 DataType::Interval(IntervalUnit::MonthDayNano)
3748 );
3749 }
3750
3751 #[test]
3752 fn test_resolve_records_mapping_default_fields_and_skip_fields() {
3753 let writer = Schema::Complex(ComplexType::Record(Record {
3754 name: "R",
3755 namespace: None,
3756 doc: None,
3757 aliases: vec![],
3758 fields: vec![
3759 crate::schema::Field {
3760 name: "a",
3761 doc: None,
3762 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3763 default: None,
3764 aliases: vec![],
3765 },
3766 crate::schema::Field {
3767 name: "skipme",
3768 doc: None,
3769 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3770 default: None,
3771 aliases: vec![],
3772 },
3773 crate::schema::Field {
3774 name: "b",
3775 doc: None,
3776 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3777 default: None,
3778 aliases: vec![],
3779 },
3780 ],
3781 attributes: Attributes::default(),
3782 }));
3783 let reader = Schema::Complex(ComplexType::Record(Record {
3784 name: "R",
3785 namespace: None,
3786 doc: None,
3787 aliases: vec![],
3788 fields: vec![
3789 crate::schema::Field {
3790 name: "b",
3791 doc: None,
3792 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3793 default: None,
3794 aliases: vec![],
3795 },
3796 crate::schema::Field {
3797 name: "a",
3798 doc: None,
3799 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3800 default: None,
3801 aliases: vec![],
3802 },
3803 crate::schema::Field {
3804 name: "name",
3805 doc: None,
3806 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3807 default: Some(json_string("anon")),
3808 aliases: vec![],
3809 },
3810 crate::schema::Field {
3811 name: "opt",
3812 doc: None,
3813 r#type: Schema::Union(vec![
3814 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
3815 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3816 ]),
3817 default: None, aliases: vec![],
3819 },
3820 ],
3821 attributes: Attributes::default(),
3822 }));
3823 let mut maker = Maker::new(false, false, Tz::default());
3824 let dt = maker
3825 .make_data_type(&writer, Some(&reader), None)
3826 .expect("record resolution");
3827 let fields = match dt.codec() {
3828 Codec::Struct(f) => f,
3829 other => panic!("expected struct, got {other:?}"),
3830 };
3831 assert_eq!(fields.len(), 4);
3832 assert_eq!(fields[0].name(), "b");
3833 assert_eq!(fields[1].name(), "a");
3834 assert_eq!(fields[2].name(), "name");
3835 assert_eq!(fields[3].name(), "opt");
3836 assert!(matches!(
3837 fields[1].data_type().resolution,
3838 Some(ResolutionInfo::Promotion(Promotion::IntToLong))
3839 ));
3840 let rec = match dt.resolution {
3841 Some(ResolutionInfo::Record(ref r)) => r.clone(),
3842 other => panic!("expected record resolution, got {other:?}"),
3843 };
3844 assert!(matches!(
3845 &rec.writer_fields[..],
3846 &[
3847 ResolvedField::ToReader(1, _),
3848 ResolvedField::Skip(_),
3849 ResolvedField::ToReader(0, _),
3850 ]
3851 ));
3852 assert_eq!(rec.default_fields.as_ref(), &[2usize, 3usize]);
3853 let ResolvedField::Skip(skip1) = &rec.writer_fields[1] else {
3854 panic!("should skip field 1")
3855 };
3856 assert!(matches!(skip1.codec(), Codec::Utf8));
3857 let name_md = &fields[2].data_type().metadata;
3858 assert_eq!(
3859 name_md.get(AVRO_FIELD_DEFAULT_METADATA_KEY),
3860 Some(&"\"anon\"".to_string())
3861 );
3862 let opt_md = &fields[3].data_type().metadata;
3863 assert_eq!(
3864 opt_md.get(AVRO_FIELD_DEFAULT_METADATA_KEY),
3865 Some(&"null".to_string())
3866 );
3867 }
3868
3869 #[test]
3870 fn test_named_type_alias_resolution_record_cross_namespace() {
3871 let writer_record = Record {
3872 name: "PersonV2",
3873 namespace: Some("com.example.v2"),
3874 doc: None,
3875 aliases: vec!["com.example.Person"],
3876 fields: vec![
3877 AvroFieldSchema {
3878 name: "name",
3879 doc: None,
3880 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3881 default: None,
3882 aliases: vec![],
3883 },
3884 AvroFieldSchema {
3885 name: "age",
3886 doc: None,
3887 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3888 default: None,
3889 aliases: vec![],
3890 },
3891 ],
3892 attributes: Attributes::default(),
3893 };
3894 let reader_record = Record {
3895 name: "Person",
3896 namespace: Some("com.example"),
3897 doc: None,
3898 aliases: vec![],
3899 fields: writer_record.fields.clone(),
3900 attributes: Attributes::default(),
3901 };
3902 let writer_schema = Schema::Complex(ComplexType::Record(writer_record));
3903 let reader_schema = Schema::Complex(ComplexType::Record(reader_record));
3904 let mut maker = Maker::new(false, false, Tz::default());
3905 let result = maker
3906 .make_data_type(&writer_schema, Some(&reader_schema), None)
3907 .expect("record alias resolution should succeed");
3908 match result.codec {
3909 Codec::Struct(ref fields) => assert_eq!(fields.len(), 2),
3910 other => panic!("expected struct, got {other:?}"),
3911 }
3912 }
3913
3914 #[test]
3915 fn test_named_type_alias_resolution_enum_cross_namespace() {
3916 let writer_enum = Enum {
3917 name: "ColorV2",
3918 namespace: Some("org.example.v2"),
3919 doc: None,
3920 aliases: vec!["org.example.Color"],
3921 symbols: vec!["RED", "GREEN", "BLUE"],
3922 default: None,
3923 attributes: Attributes::default(),
3924 };
3925 let reader_enum = Enum {
3926 name: "Color",
3927 namespace: Some("org.example"),
3928 doc: None,
3929 aliases: vec![],
3930 symbols: vec!["RED", "GREEN", "BLUE"],
3931 default: None,
3932 attributes: Attributes::default(),
3933 };
3934 let writer_schema = Schema::Complex(ComplexType::Enum(writer_enum));
3935 let reader_schema = Schema::Complex(ComplexType::Enum(reader_enum));
3936 let mut maker = Maker::new(false, false, Tz::default());
3937 maker
3938 .make_data_type(&writer_schema, Some(&reader_schema), None)
3939 .expect("enum alias resolution should succeed");
3940 }
3941
3942 #[test]
3943 fn test_named_type_alias_resolution_fixed_cross_namespace() {
3944 let writer_fixed = Fixed {
3945 name: "Fx10V2",
3946 namespace: Some("ns.v2"),
3947 aliases: vec!["ns.Fx10"],
3948 size: 10,
3949 attributes: Attributes::default(),
3950 };
3951 let reader_fixed = Fixed {
3952 name: "Fx10",
3953 namespace: Some("ns"),
3954 aliases: vec![],
3955 size: 10,
3956 attributes: Attributes::default(),
3957 };
3958 let writer_schema = Schema::Complex(ComplexType::Fixed(writer_fixed));
3959 let reader_schema = Schema::Complex(ComplexType::Fixed(reader_fixed));
3960 let mut maker = Maker::new(false, false, Tz::default());
3961 maker
3962 .make_data_type(&writer_schema, Some(&reader_schema), None)
3963 .expect("fixed alias resolution should succeed");
3964 }
3965}