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