1#[cfg(feature = "canonical_extension_types")]
21use arrow_schema::extension::ExtensionType;
22use arrow_schema::{
23 ArrowError, DataType, Field as ArrowField, IntervalUnit, Metadata, Schema as ArrowSchema,
24 TimeUnit, UnionMode,
25};
26use serde::{Deserialize, Serialize};
27use serde_json::{Map as JsonMap, Value, json};
28#[cfg(feature = "sha256")]
29use sha2::{Digest, Sha256};
30use std::borrow::Cow;
31use std::cmp::PartialEq;
32use std::collections::hash_map::Entry;
33use std::collections::{HashMap, HashSet};
34use strum_macros::AsRefStr;
35
36pub const SINGLE_OBJECT_MAGIC: [u8; 2] = [0xC3, 0x01];
38
39pub const CONFLUENT_MAGIC: [u8; 1] = [0x00];
41
42pub const MAX_PREFIX_LEN: usize = 34;
45
46pub const SCHEMA_METADATA_KEY: &str = "avro.schema";
48
49pub const AVRO_ENUM_SYMBOLS_METADATA_KEY: &str = "avro.enum.symbols";
51
52pub const AVRO_FIELD_DEFAULT_METADATA_KEY: &str = "avro.field.default";
54
55pub const AVRO_NAME_METADATA_KEY: &str = "avro.name";
57
58pub const AVRO_NAMESPACE_METADATA_KEY: &str = "avro.namespace";
60
61pub const AVRO_DOC_METADATA_KEY: &str = "avro.doc";
63
64pub const AVRO_ROOT_RECORD_DEFAULT_NAME: &str = "topLevelRecord";
66
67#[derive(Debug, Copy, Clone, PartialEq, Default)]
73pub(crate) enum Nullability {
74 #[default]
76 NullFirst,
77 NullSecond,
79}
80
81impl Nullability {
82 pub(crate) fn non_null_index(&self) -> usize {
84 match self {
85 Nullability::NullFirst => 1,
86 Nullability::NullSecond => 0,
87 }
88 }
89}
90
91#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
95#[serde(untagged)]
96pub(crate) enum TypeName<'a> {
100 Primitive(PrimitiveType),
102 Ref(&'a str),
104}
105
106#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, AsRefStr)]
110#[serde(rename_all = "camelCase")]
111#[strum(serialize_all = "lowercase")]
112pub(crate) enum PrimitiveType {
113 Null,
115 Boolean,
117 Int,
119 Long,
121 Float,
123 Double,
125 Bytes,
127 String,
129}
130
131#[derive(Debug, Clone, PartialEq, Eq, Default, Deserialize, Serialize)]
135#[serde(rename_all = "camelCase")]
136pub(crate) struct Attributes<'a> {
137 #[serde(default)]
141 pub(crate) logical_type: Option<&'a str>,
142
143 #[serde(flatten)]
145 pub(crate) additional: HashMap<&'a str, Value>,
146}
147
148impl Attributes<'_> {
149 pub(crate) fn field_metadata(&self) -> HashMap<String, String> {
151 self.additional
152 .iter()
153 .map(|(k, v)| (k.to_string(), v.to_string()))
154 .collect()
155 }
156}
157
158#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
160#[serde(rename_all = "camelCase")]
161pub(crate) struct Type<'a> {
162 #[serde(borrow)]
164 pub(crate) r#type: TypeName<'a>,
165 #[serde(flatten)]
167 pub(crate) attributes: Attributes<'a>,
168}
169
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
175#[serde(untagged)]
176pub(crate) enum Schema<'a> {
177 #[serde(borrow)]
179 TypeName(TypeName<'a>),
180 #[serde(borrow)]
182 Union(Vec<Schema<'a>>),
183 #[serde(borrow)]
185 Complex(ComplexType<'a>),
186 #[serde(borrow)]
188 Type(Type<'a>),
189}
190
191#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
195#[serde(tag = "type", rename_all = "camelCase")]
196pub(crate) enum ComplexType<'a> {
197 #[serde(borrow)]
199 Record(Record<'a>),
200 #[serde(borrow)]
202 Enum(Enum<'a>),
203 #[serde(borrow)]
205 Array(Array<'a>),
206 #[serde(borrow)]
208 Map(Map<'a>),
209 #[serde(borrow)]
211 Fixed(Fixed<'a>),
212}
213
214#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
218pub(crate) struct Record<'a> {
219 #[serde(borrow)]
221 pub(crate) name: &'a str,
222 #[serde(borrow, default)]
224 pub(crate) namespace: Option<&'a str>,
225 #[serde(borrow, default)]
227 pub(crate) doc: Option<Cow<'a, str>>,
228 #[serde(borrow, default)]
230 pub(crate) aliases: Vec<&'a str>,
231 #[serde(borrow)]
233 pub(crate) fields: Vec<Field<'a>>,
234 #[serde(flatten)]
236 pub(crate) attributes: Attributes<'a>,
237}
238
239fn deserialize_default<'de, D>(deserializer: D) -> Result<Option<Value>, D::Error>
240where
241 D: serde::Deserializer<'de>,
242{
243 Value::deserialize(deserializer).map(Some)
244}
245
246#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
248pub(crate) struct Field<'a> {
249 #[serde(borrow)]
251 pub(crate) name: &'a str,
252 #[serde(borrow, default)]
254 pub(crate) doc: Option<Cow<'a, str>>,
255 #[serde(borrow)]
257 pub(crate) r#type: Schema<'a>,
258 #[serde(deserialize_with = "deserialize_default", default)]
260 pub(crate) default: Option<Value>,
261 #[serde(borrow, default)]
264 pub(crate) aliases: Vec<&'a str>,
265}
266
267#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
271pub(crate) struct Enum<'a> {
272 #[serde(borrow)]
274 pub(crate) name: &'a str,
275 #[serde(borrow, default)]
277 pub(crate) namespace: Option<&'a str>,
278 #[serde(borrow, default)]
280 pub(crate) doc: Option<Cow<'a, str>>,
281 #[serde(borrow, default)]
283 pub(crate) aliases: Vec<&'a str>,
284 #[serde(borrow)]
286 pub(crate) symbols: Vec<&'a str>,
287 #[serde(borrow, default)]
289 pub(crate) default: Option<&'a str>,
290 #[serde(flatten)]
292 pub(crate) attributes: Attributes<'a>,
293}
294
295#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
299pub(crate) struct Array<'a> {
300 #[serde(borrow)]
302 pub(crate) items: Box<Schema<'a>>,
303 #[serde(flatten)]
305 pub(crate) attributes: Attributes<'a>,
306}
307
308#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
312pub(crate) struct Map<'a> {
313 #[serde(borrow)]
315 pub(crate) values: Box<Schema<'a>>,
316 #[serde(flatten)]
318 pub(crate) attributes: Attributes<'a>,
319}
320
321#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
325pub(crate) struct Fixed<'a> {
326 #[serde(borrow)]
328 pub(crate) name: &'a str,
329 #[serde(borrow, default)]
331 pub(crate) namespace: Option<&'a str>,
332 #[serde(borrow, default)]
334 pub(crate) aliases: Vec<&'a str>,
335 pub(crate) size: usize,
337 #[serde(flatten)]
339 pub(crate) attributes: Attributes<'a>,
340}
341
342#[derive(Debug, Copy, Clone, PartialEq, Default)]
343pub(crate) struct AvroSchemaOptions {
344 pub(crate) null_order: Option<Nullability>,
345 pub(crate) strip_metadata: bool,
346}
347
348#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
350pub struct AvroSchema {
351 pub json_string: String,
353}
354
355impl TryFrom<&ArrowSchema> for AvroSchema {
356 type Error = ArrowError;
357
358 fn try_from(schema: &ArrowSchema) -> Result<Self, Self::Error> {
362 AvroSchema::from_arrow_with_options(schema, None)
363 }
364}
365
366impl AvroSchema {
367 pub fn new(json_string: String) -> Self {
369 Self { json_string }
370 }
371
372 pub(crate) fn schema(&self) -> Result<Schema<'_>, ArrowError> {
373 serde_json::from_str(self.json_string.as_str())
374 .map_err(|e| ArrowError::ParseError(format!("Invalid Avro schema JSON: {e}")))
375 }
376
377 pub fn fingerprint(&self, hash_type: FingerprintAlgorithm) -> Result<Fingerprint, ArrowError> {
405 Self::generate_fingerprint(&self.schema()?, hash_type)
406 }
407
408 pub(crate) fn project(&self, projection: &[usize]) -> Result<Self, ArrowError> {
409 let mut value: Value = serde_json::from_str(&self.json_string)
410 .map_err(|e| ArrowError::AvroError(format!("Invalid Avro schema JSON: {e}")))?;
411 let obj = value.as_object_mut().ok_or_else(|| {
412 ArrowError::AvroError(
413 "Projected schema must be a JSON object Avro record schema".to_string(),
414 )
415 })?;
416 match obj.get("type").and_then(|v| v.as_str()) {
417 Some("record") => {}
418 Some(other) => {
419 return Err(ArrowError::AvroError(format!(
420 "Projected schema must be an Avro record, found type '{other}'"
421 )));
422 }
423 None => {
424 return Err(ArrowError::AvroError(
425 "Projected schema missing required 'type' field".to_string(),
426 ));
427 }
428 }
429 let fields_val = obj.get_mut("fields").ok_or_else(|| {
430 ArrowError::AvroError("Avro record schema missing required 'fields'".to_string())
431 })?;
432 let projected_fields = {
433 let mut original_fields = match fields_val {
434 Value::Array(arr) => std::mem::take(arr),
435 _ => {
436 return Err(ArrowError::AvroError(
437 "Avro record schema 'fields' must be an array".to_string(),
438 ));
439 }
440 };
441 let len = original_fields.len();
442 let mut seen: HashSet<usize> = HashSet::with_capacity(projection.len());
443 let mut out: Vec<Value> = Vec::with_capacity(projection.len());
444 for &i in projection {
445 if i >= len {
446 return Err(ArrowError::AvroError(format!(
447 "Projection index {i} out of bounds for record with {len} fields"
448 )));
449 }
450 if !seen.insert(i) {
451 return Err(ArrowError::AvroError(format!(
452 "Duplicate projection index {i}"
453 )));
454 }
455 out.push(std::mem::replace(&mut original_fields[i], Value::Null));
456 }
457 out
458 };
459 *fields_val = Value::Array(projected_fields);
460 let json_string = serde_json::to_string(&value).map_err(|e| {
461 ArrowError::AvroError(format!(
462 "Failed to serialize projected Avro schema JSON: {e}"
463 ))
464 })?;
465 Ok(Self::new(json_string))
466 }
467
468 pub(crate) fn generate_fingerprint(
469 schema: &Schema,
470 hash_type: FingerprintAlgorithm,
471 ) -> Result<Fingerprint, ArrowError> {
472 let canonical = Self::generate_canonical_form(schema).map_err(|e| {
473 ArrowError::ComputeError(format!("Failed to generate canonical form for schema: {e}"))
474 })?;
475 match hash_type {
476 FingerprintAlgorithm::Rabin => {
477 Ok(Fingerprint::Rabin(compute_fingerprint_rabin(&canonical)))
478 }
479 FingerprintAlgorithm::Id | FingerprintAlgorithm::Id64 => Err(ArrowError::SchemaError(
480 "FingerprintAlgorithm of Id or Id64 cannot be used to generate a fingerprint; \
481 if using Fingerprint::Id, pass the registry ID in instead using the set method."
482 .to_string(),
483 )),
484 #[cfg(feature = "md5")]
485 FingerprintAlgorithm::MD5 => Ok(Fingerprint::MD5(compute_fingerprint_md5(&canonical))),
486 #[cfg(feature = "sha256")]
487 FingerprintAlgorithm::SHA256 => {
488 Ok(Fingerprint::SHA256(compute_fingerprint_sha256(&canonical)))
489 }
490 }
491 }
492
493 pub(crate) fn generate_canonical_form(schema: &Schema) -> Result<String, ArrowError> {
504 build_canonical(schema, None)
505 }
506
507 pub(crate) fn from_arrow_with_options(
514 schema: &ArrowSchema,
515 options: Option<AvroSchemaOptions>,
516 ) -> Result<AvroSchema, ArrowError> {
517 let opts = options.unwrap_or_default();
518 let order = opts.null_order.unwrap_or_default();
519 let strip = opts.strip_metadata;
520 if !strip && let Some(json) = schema.metadata.get(SCHEMA_METADATA_KEY) {
521 return Ok(AvroSchema::new(json.clone()));
522 }
523 let mut name_gen = NameGenerator::default();
524 let fields_json = schema
525 .fields()
526 .iter()
527 .map(|f| arrow_field_to_avro(f, &mut name_gen, order, strip))
528 .collect::<Result<Vec<_>, _>>()?;
529 let record_name = schema
530 .metadata
531 .get(AVRO_NAME_METADATA_KEY)
532 .map_or(AVRO_ROOT_RECORD_DEFAULT_NAME, |s| s.as_str());
533 let mut record = JsonMap::with_capacity(schema.metadata.len() + 4);
534 record.insert("type".into(), Value::String("record".into()));
535 record.insert(
536 "name".into(),
537 Value::String(sanitise_avro_name(record_name)),
538 );
539 if let Some(ns) = schema.metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
540 record.insert("namespace".into(), Value::String(ns.clone()));
541 }
542 if let Some(doc) = schema.metadata.get(AVRO_DOC_METADATA_KEY) {
543 record.insert("doc".into(), Value::String(doc.clone()));
544 }
545 record.insert("fields".into(), Value::Array(fields_json));
546 extend_with_passthrough_metadata(&mut record, &schema.metadata);
547 let json_string = serde_json::to_string(&Value::Object(record))
548 .map_err(|e| ArrowError::SchemaError(format!("Serializing Avro JSON failed: {e}")))?;
549 Ok(AvroSchema::new(json_string))
550 }
551}
552
553#[derive(Debug, Copy, Clone)]
555pub(crate) struct Prefix {
556 buf: [u8; MAX_PREFIX_LEN],
557 len: u8,
558}
559
560impl Prefix {
561 #[inline]
562 pub(crate) fn as_slice(&self) -> &[u8] {
563 &self.buf[..self.len as usize]
564 }
565}
566
567#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
569pub enum FingerprintStrategy {
570 #[default]
572 Rabin,
573 Id(u32),
575 Id64(u64),
577 #[cfg(feature = "md5")]
578 MD5,
580 #[cfg(feature = "sha256")]
581 SHA256,
583}
584
585impl From<Fingerprint> for FingerprintStrategy {
586 fn from(f: Fingerprint) -> Self {
587 Self::from(&f)
588 }
589}
590
591impl From<FingerprintAlgorithm> for FingerprintStrategy {
592 fn from(f: FingerprintAlgorithm) -> Self {
593 match f {
594 FingerprintAlgorithm::Rabin => FingerprintStrategy::Rabin,
595 FingerprintAlgorithm::Id => FingerprintStrategy::Id(0),
596 FingerprintAlgorithm::Id64 => FingerprintStrategy::Id64(0),
597 #[cfg(feature = "md5")]
598 FingerprintAlgorithm::MD5 => FingerprintStrategy::MD5,
599 #[cfg(feature = "sha256")]
600 FingerprintAlgorithm::SHA256 => FingerprintStrategy::SHA256,
601 }
602 }
603}
604
605impl From<&Fingerprint> for FingerprintStrategy {
606 fn from(f: &Fingerprint) -> Self {
607 match f {
608 Fingerprint::Rabin(_) => FingerprintStrategy::Rabin,
609 Fingerprint::Id(_) => FingerprintStrategy::Id(0),
610 Fingerprint::Id64(_) => FingerprintStrategy::Id64(0),
611 #[cfg(feature = "md5")]
612 Fingerprint::MD5(_) => FingerprintStrategy::MD5,
613 #[cfg(feature = "sha256")]
614 Fingerprint::SHA256(_) => FingerprintStrategy::SHA256,
615 }
616 }
617}
618
619#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Default)]
622pub enum FingerprintAlgorithm {
623 #[default]
625 Rabin,
626 Id,
628 Id64,
630 #[cfg(feature = "md5")]
631 MD5,
633 #[cfg(feature = "sha256")]
634 SHA256,
636}
637
638impl From<&Fingerprint> for FingerprintAlgorithm {
640 fn from(fp: &Fingerprint) -> Self {
641 match fp {
642 Fingerprint::Rabin(_) => FingerprintAlgorithm::Rabin,
643 Fingerprint::Id(_) => FingerprintAlgorithm::Id,
644 Fingerprint::Id64(_) => FingerprintAlgorithm::Id64,
645 #[cfg(feature = "md5")]
646 Fingerprint::MD5(_) => FingerprintAlgorithm::MD5,
647 #[cfg(feature = "sha256")]
648 Fingerprint::SHA256(_) => FingerprintAlgorithm::SHA256,
649 }
650 }
651}
652
653impl From<FingerprintStrategy> for FingerprintAlgorithm {
654 fn from(s: FingerprintStrategy) -> Self {
655 Self::from(&s)
656 }
657}
658
659impl From<&FingerprintStrategy> for FingerprintAlgorithm {
660 fn from(s: &FingerprintStrategy) -> Self {
661 match s {
662 FingerprintStrategy::Rabin => FingerprintAlgorithm::Rabin,
663 FingerprintStrategy::Id(_) => FingerprintAlgorithm::Id,
664 FingerprintStrategy::Id64(_) => FingerprintAlgorithm::Id64,
665 #[cfg(feature = "md5")]
666 FingerprintStrategy::MD5 => FingerprintAlgorithm::MD5,
667 #[cfg(feature = "sha256")]
668 FingerprintStrategy::SHA256 => FingerprintAlgorithm::SHA256,
669 }
670 }
671}
672
673#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
682pub enum Fingerprint {
683 Rabin(u64),
685 Id(u32),
687 Id64(u64),
689 #[cfg(feature = "md5")]
690 MD5([u8; 16]),
692 #[cfg(feature = "sha256")]
693 SHA256([u8; 32]),
695}
696
697impl From<FingerprintStrategy> for Fingerprint {
698 fn from(s: FingerprintStrategy) -> Self {
699 Self::from(&s)
700 }
701}
702
703impl From<&FingerprintStrategy> for Fingerprint {
704 fn from(s: &FingerprintStrategy) -> Self {
705 match s {
706 FingerprintStrategy::Rabin => Fingerprint::Rabin(0),
707 FingerprintStrategy::Id(id) => Fingerprint::Id(*id),
708 FingerprintStrategy::Id64(id) => Fingerprint::Id64(*id),
709 #[cfg(feature = "md5")]
710 FingerprintStrategy::MD5 => Fingerprint::MD5([0; 16]),
711 #[cfg(feature = "sha256")]
712 FingerprintStrategy::SHA256 => Fingerprint::SHA256([0; 32]),
713 }
714 }
715}
716
717impl From<FingerprintAlgorithm> for Fingerprint {
718 fn from(s: FingerprintAlgorithm) -> Self {
719 match s {
720 FingerprintAlgorithm::Rabin => Fingerprint::Rabin(0),
721 FingerprintAlgorithm::Id => Fingerprint::Id(0),
722 FingerprintAlgorithm::Id64 => Fingerprint::Id64(0),
723 #[cfg(feature = "md5")]
724 FingerprintAlgorithm::MD5 => Fingerprint::MD5([0; 16]),
725 #[cfg(feature = "sha256")]
726 FingerprintAlgorithm::SHA256 => Fingerprint::SHA256([0; 32]),
727 }
728 }
729}
730
731impl Fingerprint {
732 pub fn load_fingerprint_id(id: u32) -> Self {
740 Fingerprint::Id(u32::from_be(id))
741 }
742
743 pub fn load_fingerprint_id64(id: u64) -> Self {
751 Fingerprint::Id64(u64::from_be(id))
752 }
753
754 pub(crate) fn make_prefix(&self) -> Prefix {
778 let mut buf = [0u8; MAX_PREFIX_LEN];
779 let len = match self {
780 Self::Id(val) => write_prefix(&mut buf, &CONFLUENT_MAGIC, &val.to_be_bytes()),
781 Self::Id64(val) => write_prefix(&mut buf, &CONFLUENT_MAGIC, &val.to_be_bytes()),
782 Self::Rabin(val) => write_prefix(&mut buf, &SINGLE_OBJECT_MAGIC, &val.to_le_bytes()),
783 #[cfg(feature = "md5")]
784 Self::MD5(val) => write_prefix(&mut buf, &SINGLE_OBJECT_MAGIC, val),
785 #[cfg(feature = "sha256")]
786 Self::SHA256(val) => write_prefix(&mut buf, &SINGLE_OBJECT_MAGIC, val),
787 };
788 Prefix { buf, len }
789 }
790}
791
792fn write_prefix<const MAGIC_LEN: usize, const PAYLOAD_LEN: usize>(
793 buf: &mut [u8; MAX_PREFIX_LEN],
794 magic: &[u8; MAGIC_LEN],
795 payload: &[u8; PAYLOAD_LEN],
796) -> u8 {
797 debug_assert!(MAGIC_LEN + PAYLOAD_LEN <= MAX_PREFIX_LEN);
798 let total = MAGIC_LEN + PAYLOAD_LEN;
799 let prefix_slice = &mut buf[..total];
800 prefix_slice[..MAGIC_LEN].copy_from_slice(magic);
801 prefix_slice[MAGIC_LEN..total].copy_from_slice(payload);
802 total as u8
803}
804
805#[derive(Debug, Clone, Default)]
832pub struct SchemaStore {
833 fingerprint_algorithm: FingerprintAlgorithm,
835 schemas: HashMap<Fingerprint, AvroSchema>,
837}
838
839impl TryFrom<HashMap<Fingerprint, AvroSchema>> for SchemaStore {
840 type Error = ArrowError;
841
842 fn try_from(schemas: HashMap<Fingerprint, AvroSchema>) -> Result<Self, Self::Error> {
845 Ok(Self {
846 schemas,
847 ..Self::default()
848 })
849 }
850}
851
852impl SchemaStore {
853 pub fn new() -> Self {
855 Self::default()
856 }
857
858 pub fn new_with_type(fingerprint_algorithm: FingerprintAlgorithm) -> Self {
860 Self {
861 fingerprint_algorithm,
862 ..Self::default()
863 }
864 }
865
866 pub fn set(
883 &mut self,
884 fingerprint: Fingerprint,
885 schema: AvroSchema,
886 ) -> Result<Fingerprint, ArrowError> {
887 match self.schemas.entry(fingerprint) {
888 Entry::Occupied(entry) => {
889 if entry.get() != &schema {
890 return Err(ArrowError::ComputeError(format!(
891 "Schema fingerprint collision detected for fingerprint {fingerprint:?}"
892 )));
893 }
894 }
895 Entry::Vacant(entry) => {
896 entry.insert(schema);
897 }
898 }
899 Ok(fingerprint)
900 }
901
902 pub fn register(&mut self, schema: AvroSchema) -> Result<Fingerprint, ArrowError> {
920 if self.fingerprint_algorithm == FingerprintAlgorithm::Id
921 || self.fingerprint_algorithm == FingerprintAlgorithm::Id64
922 {
923 return Err(ArrowError::SchemaError(
924 "Invalid FingerprintAlgorithm; unable to generate fingerprint. \
925 Use the set method directly instead, providing a valid fingerprint"
926 .to_string(),
927 ));
928 }
929 let fingerprint =
930 AvroSchema::generate_fingerprint(&schema.schema()?, self.fingerprint_algorithm)?;
931 self.set(fingerprint, schema)?;
932 Ok(fingerprint)
933 }
934
935 pub fn lookup(&self, fingerprint: &Fingerprint) -> Option<&AvroSchema> {
945 self.schemas.get(fingerprint)
946 }
947
948 pub fn fingerprints(&self) -> Vec<Fingerprint> {
954 self.schemas.keys().copied().collect()
955 }
956
957 pub(crate) fn fingerprint_algorithm(&self) -> FingerprintAlgorithm {
959 self.fingerprint_algorithm
960 }
961}
962
963fn quote(s: &str) -> Result<String, ArrowError> {
964 serde_json::to_string(s)
965 .map_err(|e| ArrowError::ComputeError(format!("Failed to quote string: {e}")))
966}
967
968pub(crate) fn make_full_name(
985 name: &str,
986 namespace_attr: Option<&str>,
987 enclosing_ns: Option<&str>,
988) -> (String, Option<String>) {
989 if let Some((ns, _)) = name.rsplit_once('.') {
991 return (name.to_string(), Some(ns.to_string()));
992 }
993 match namespace_attr.or(enclosing_ns) {
994 Some(ns) => (format!("{ns}.{name}"), Some(ns.to_string())),
995 None => (name.to_string(), None),
996 }
997}
998
999fn build_canonical(schema: &Schema, enclosing_ns: Option<&str>) -> Result<String, ArrowError> {
1000 Ok(match schema {
1001 Schema::TypeName(tn) | Schema::Type(Type { r#type: tn, .. }) => match tn {
1002 TypeName::Primitive(pt) => quote(pt.as_ref())?,
1003 TypeName::Ref(name) => {
1004 let (full_name, _) = make_full_name(name, None, enclosing_ns);
1005 quote(&full_name)?
1006 }
1007 },
1008 Schema::Union(branches) => format!(
1009 "[{}]",
1010 branches
1011 .iter()
1012 .map(|b| build_canonical(b, enclosing_ns))
1013 .collect::<Result<Vec<_>, _>>()?
1014 .join(",")
1015 ),
1016 Schema::Complex(ct) => match ct {
1017 ComplexType::Record(r) => {
1018 let (full_name, child_ns) = make_full_name(r.name, r.namespace, enclosing_ns);
1019 let fields = r
1020 .fields
1021 .iter()
1022 .map(|f| {
1023 let field_type =
1028 build_canonical(&f.r#type, child_ns.as_deref().or(enclosing_ns))?;
1029 Ok(format!(
1030 r#"{{"name":{},"type":{}}}"#,
1031 quote(f.name)?,
1032 field_type
1033 ))
1034 })
1035 .collect::<Result<Vec<_>, ArrowError>>()?
1036 .join(",");
1037 format!(
1038 r#"{{"name":{},"type":"record","fields":[{fields}]}}"#,
1039 quote(&full_name)?,
1040 )
1041 }
1042 ComplexType::Enum(e) => {
1043 let (full_name, _) = make_full_name(e.name, e.namespace, enclosing_ns);
1044 let symbols = e
1045 .symbols
1046 .iter()
1047 .map(|s| quote(s))
1048 .collect::<Result<Vec<_>, _>>()?
1049 .join(",");
1050 format!(
1051 r#"{{"name":{},"type":"enum","symbols":[{symbols}]}}"#,
1052 quote(&full_name)?
1053 )
1054 }
1055 ComplexType::Array(arr) => format!(
1056 r#"{{"type":"array","items":{}}}"#,
1057 build_canonical(&arr.items, enclosing_ns)?
1058 ),
1059 ComplexType::Map(map) => format!(
1060 r#"{{"type":"map","values":{}}}"#,
1061 build_canonical(&map.values, enclosing_ns)?
1062 ),
1063 ComplexType::Fixed(f) => {
1064 let (full_name, _) = make_full_name(f.name, f.namespace, enclosing_ns);
1065 format!(
1066 r#"{{"name":{},"type":"fixed","size":{}}}"#,
1067 quote(&full_name)?,
1068 f.size
1069 )
1070 }
1071 },
1072 })
1073}
1074
1075const EMPTY: u64 = 0xc15d_213a_a4d7_a795;
1077
1078const fn one_entry(i: usize) -> u64 {
1085 let mut fp = i as u64;
1086 let mut j = 0;
1087 while j < 8 {
1088 fp = (fp >> 1) ^ (EMPTY & (0u64.wrapping_sub(fp & 1)));
1089 j += 1;
1090 }
1091 fp
1092}
1093
1094const fn build_table() -> [u64; 256] {
1101 let mut table = [0u64; 256];
1102 let mut i = 0;
1103 while i < 256 {
1104 table[i] = one_entry(i);
1105 i += 1;
1106 }
1107 table
1108}
1109
1110static FINGERPRINT_TABLE: [u64; 256] = build_table();
1112
1113pub(crate) fn compute_fingerprint_rabin(canonical_form: &str) -> u64 {
1116 let mut fp = EMPTY;
1117 for &byte in canonical_form.as_bytes() {
1118 let idx = ((fp as u8) ^ byte) as usize;
1119 fp = (fp >> 8) ^ FINGERPRINT_TABLE[idx];
1120 }
1121 fp
1122}
1123
1124#[cfg(feature = "md5")]
1125#[inline]
1130pub(crate) fn compute_fingerprint_md5(canonical_form: &str) -> [u8; 16] {
1131 let digest = md5::compute(canonical_form.as_bytes());
1132 digest.0
1133}
1134
1135#[cfg(feature = "sha256")]
1136#[inline]
1140pub(crate) fn compute_fingerprint_sha256(canonical_form: &str) -> [u8; 32] {
1141 let mut hasher = Sha256::new();
1142 hasher.update(canonical_form.as_bytes());
1143 let digest = hasher.finalize();
1144 digest.into()
1145}
1146
1147#[inline]
1148fn is_internal_arrow_key(key: &str) -> bool {
1149 key.starts_with("ARROW:") || key == SCHEMA_METADATA_KEY
1150}
1151
1152fn extend_with_passthrough_metadata(target: &mut JsonMap<String, Value>, metadata: &Metadata) {
1157 for (meta_key, meta_val) in metadata {
1158 if meta_key.starts_with("avro.") || is_internal_arrow_key(meta_key) {
1159 continue;
1160 }
1161 let json_val =
1162 serde_json::from_str(meta_val).unwrap_or_else(|_| Value::String(meta_val.clone()));
1163 target.insert(meta_key.clone(), json_val);
1164 }
1165}
1166
1167fn sanitise_avro_name(base_name: &str) -> String {
1169 if base_name.is_empty() {
1170 return "_".to_owned();
1171 }
1172 let mut out: String = base_name
1173 .chars()
1174 .map(|char| {
1175 if char.is_ascii_alphanumeric() || char == '_' {
1176 char
1177 } else {
1178 '_'
1179 }
1180 })
1181 .collect();
1182 if out.as_bytes()[0].is_ascii_digit() {
1183 out.insert(0, '_');
1184 }
1185 out
1186}
1187
1188#[derive(Default)]
1189struct NameGenerator {
1190 used: HashSet<String>,
1191 counters: HashMap<String, usize>,
1192}
1193
1194impl NameGenerator {
1195 fn make_unique(&mut self, field_name: &str) -> String {
1196 let field_name = sanitise_avro_name(field_name);
1197 if self.used.insert(field_name.clone()) {
1198 self.counters.insert(field_name.clone(), 1);
1199 return field_name;
1200 }
1201 let counter = self.counters.entry(field_name.clone()).or_insert(1);
1202 loop {
1203 let candidate = format!("{field_name}_{}", *counter);
1204 if self.used.insert(candidate.clone()) {
1205 return candidate;
1206 }
1207 *counter += 1;
1208 }
1209 }
1210}
1211
1212fn merge_extras(schema: Value, extras: JsonMap<String, Value>) -> Value {
1213 if extras.is_empty() {
1214 return schema;
1215 }
1216 match schema {
1217 Value::Object(mut map) => {
1218 map.extend(extras);
1219 Value::Object(map)
1220 }
1221 Value::Array(mut union) => {
1222 if let Some(non_null) = union.iter_mut().find(|val| val.as_str() != Some("null")) {
1225 let original = std::mem::take(non_null);
1226 *non_null = merge_extras(original, extras);
1227 }
1228 Value::Array(union)
1229 }
1230 primitive => {
1231 let mut map = JsonMap::with_capacity(extras.len() + 1);
1232 map.insert("type".into(), primitive);
1233 map.extend(extras);
1234 Value::Object(map)
1235 }
1236 }
1237}
1238
1239#[inline]
1240fn is_avro_json_null(v: &Value) -> bool {
1241 matches!(v, Value::String(s) if s == "null")
1242}
1243
1244fn wrap_nullable(inner: Value, null_order: Nullability) -> Value {
1245 let null = Value::String("null".into());
1246 match inner {
1247 Value::Array(mut union) => {
1248 if union.iter().any(is_avro_json_null) {
1253 return Value::Array(union);
1254 }
1255 match null_order {
1257 Nullability::NullFirst => union.insert(0, null),
1258 Nullability::NullSecond => union.push(null),
1259 }
1260 Value::Array(union)
1261 }
1262 other => match null_order {
1263 Nullability::NullFirst => Value::Array(vec![null, other]),
1264 Nullability::NullSecond => Value::Array(vec![other, null]),
1265 },
1266 }
1267}
1268
1269fn min_fixed_bytes_for_precision(p: usize) -> usize {
1270 const MAX_P: [usize; 32] = [
1273 2, 4, 6, 9, 11, 14, 16, 18, 21, 23, 26, 28, 31, 33, 35, 38, 40, 43, 45, 47, 50, 52, 55, 57,
1274 59, 62, 64, 67, 69, 71, 74, 76,
1275 ];
1276 for (i, &max_p) in MAX_P.iter().enumerate() {
1277 if p <= max_p {
1278 return i + 1;
1279 }
1280 }
1281 32 }
1283
1284fn union_branch_signature(branch: &Value) -> Result<String, ArrowError> {
1285 match branch {
1286 Value::String(t) => Ok(format!("P:{t}")),
1287 Value::Object(map) => {
1288 let t = map.get("type").and_then(|v| v.as_str()).ok_or_else(|| {
1289 ArrowError::SchemaError("Union branch object missing string 'type'".into())
1290 })?;
1291 match t {
1292 "record" | "enum" | "fixed" => {
1293 let name = map.get("name").and_then(|v| v.as_str()).ok_or_else(|| {
1294 ArrowError::SchemaError(format!(
1295 "Union branch '{t}' missing required 'name'"
1296 ))
1297 })?;
1298 Ok(format!("N:{t}:{name}"))
1299 }
1300 "array" | "map" => Ok(format!("C:{t}")),
1301 other => Ok(format!("P:{other}")),
1302 }
1303 }
1304 Value::Array(_) => Err(ArrowError::SchemaError(
1305 "Avro union may not immediately contain another union".into(),
1306 )),
1307 _ => Err(ArrowError::SchemaError(
1308 "Invalid JSON for Avro union branch".into(),
1309 )),
1310 }
1311}
1312
1313fn datatype_to_avro(
1314 dt: &DataType,
1315 field_name: &str,
1316 metadata: &Metadata,
1317 name_gen: &mut NameGenerator,
1318 null_order: Nullability,
1319 strip: bool,
1320) -> Result<(Value, JsonMap<String, Value>), ArrowError> {
1321 let mut extras = JsonMap::new();
1322 let mut handle_decimal = |precision: &u8, scale: &i8| -> Result<Value, ArrowError> {
1323 if *scale < 0 {
1324 return Err(ArrowError::SchemaError(format!(
1325 "Invalid Avro decimal for field '{field_name}': scale ({scale}) must be >= 0"
1326 )));
1327 }
1328 if (*scale as usize) > (*precision as usize) {
1329 return Err(ArrowError::SchemaError(format!(
1330 "Invalid Avro decimal for field '{field_name}': scale ({scale}) \
1331 must be <= precision ({precision})"
1332 )));
1333 }
1334 let mut meta = JsonMap::from_iter([
1335 ("logicalType".into(), json!("decimal")),
1336 ("precision".into(), json!(*precision)),
1337 ("scale".into(), json!(*scale)),
1338 ]);
1339 let mut fixed_size = metadata.get("size").and_then(|v| v.parse::<usize>().ok());
1340 let carries_name = metadata.contains_key(AVRO_NAME_METADATA_KEY)
1341 || metadata.contains_key(AVRO_NAMESPACE_METADATA_KEY);
1342 if fixed_size.is_none() && carries_name {
1343 fixed_size = Some(min_fixed_bytes_for_precision(*precision as usize));
1344 }
1345 if let Some(size) = fixed_size {
1346 meta.insert("type".into(), json!("fixed"));
1347 meta.insert("size".into(), json!(size));
1348 let chosen_name = metadata
1349 .get(AVRO_NAME_METADATA_KEY)
1350 .map(|s| sanitise_avro_name(s))
1351 .unwrap_or_else(|| name_gen.make_unique(field_name));
1352 meta.insert("name".into(), json!(chosen_name));
1353 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1354 meta.insert("namespace".into(), json!(ns));
1355 }
1356 } else {
1357 meta.insert("type".into(), json!("bytes"));
1359 }
1360 Ok(Value::Object(meta))
1361 };
1362 let val = match dt {
1363 DataType::Null => Value::String("null".into()),
1364 DataType::Boolean => Value::String("boolean".into()),
1365 #[cfg(not(feature = "avro_custom_types"))]
1366 DataType::Int8 | DataType::Int16 | DataType::UInt8 | DataType::UInt16 => {
1367 Value::String("int".into())
1368 }
1369 DataType::Int32 => Value::String("int".into()),
1370 #[cfg(feature = "avro_custom_types")]
1371 DataType::Int8 => json!({ "type": "int", "logicalType": "arrow.int8" }),
1372 #[cfg(feature = "avro_custom_types")]
1373 DataType::Int16 => json!({ "type": "int", "logicalType": "arrow.int16" }),
1374 #[cfg(feature = "avro_custom_types")]
1375 DataType::UInt8 => json!({ "type": "int", "logicalType": "arrow.uint8" }),
1376 #[cfg(feature = "avro_custom_types")]
1377 DataType::UInt16 => json!({ "type": "int", "logicalType": "arrow.uint16" }),
1378 #[cfg(not(feature = "avro_custom_types"))]
1379 DataType::UInt32 => Value::String("long".into()),
1380 #[cfg(feature = "avro_custom_types")]
1381 DataType::UInt32 => json!({ "type": "long", "logicalType": "arrow.uint32" }),
1382 DataType::Int64 => Value::String("long".into()),
1383 #[cfg(not(feature = "avro_custom_types"))]
1384 DataType::UInt64 => Value::String("long".into()),
1385 #[cfg(feature = "avro_custom_types")]
1386 DataType::UInt64 => {
1387 let chosen_name = metadata
1389 .get(AVRO_NAME_METADATA_KEY)
1390 .map(|s| sanitise_avro_name(s))
1391 .unwrap_or_else(|| name_gen.make_unique(field_name));
1392 let mut obj = JsonMap::from_iter([
1393 ("type".into(), json!("fixed")),
1394 ("name".into(), json!(chosen_name)),
1395 ("size".into(), json!(8)),
1396 ("logicalType".into(), json!("arrow.uint64")),
1397 ]);
1398 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1399 obj.insert("namespace".into(), json!(ns));
1400 }
1401 json!(obj)
1402 }
1403 #[cfg(not(feature = "avro_custom_types"))]
1404 DataType::Float16 => Value::String("float".into()),
1405 #[cfg(feature = "avro_custom_types")]
1406 DataType::Float16 => {
1407 let chosen_name = metadata
1409 .get(AVRO_NAME_METADATA_KEY)
1410 .map(|s| sanitise_avro_name(s))
1411 .unwrap_or_else(|| name_gen.make_unique(field_name));
1412 let mut obj = JsonMap::from_iter([
1413 ("type".into(), json!("fixed")),
1414 ("name".into(), json!(chosen_name)),
1415 ("size".into(), json!(2)),
1416 ("logicalType".into(), json!("arrow.float16")),
1417 ]);
1418 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1419 obj.insert("namespace".into(), json!(ns));
1420 }
1421 json!(obj)
1422 }
1423 DataType::Float32 => Value::String("float".into()),
1424 DataType::Float64 => Value::String("double".into()),
1425 DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => Value::String("string".into()),
1426 DataType::Binary | DataType::LargeBinary => Value::String("bytes".into()),
1427 DataType::BinaryView => {
1428 if !strip {
1429 extras.insert("arrowBinaryView".into(), Value::Bool(true));
1430 }
1431 Value::String("bytes".into())
1432 }
1433 DataType::FixedSizeBinary(len) => {
1434 let md_is_uuid = metadata
1435 .get("logicalType")
1436 .map(|s| s.trim_matches('"') == "uuid")
1437 .unwrap_or(false);
1438 #[cfg(feature = "canonical_extension_types")]
1439 let ext_is_uuid = metadata
1440 .get(arrow_schema::extension::EXTENSION_TYPE_NAME_KEY)
1441 .map(|v| v == arrow_schema::extension::Uuid::NAME || v == "uuid")
1442 .unwrap_or(false);
1443 #[cfg(not(feature = "canonical_extension_types"))]
1444 let ext_is_uuid = false;
1445 let is_uuid = (*len == 16) && (md_is_uuid || ext_is_uuid);
1446 if is_uuid {
1447 json!({ "type": "string", "logicalType": "uuid" })
1448 } else {
1449 let chosen_name = metadata
1450 .get(AVRO_NAME_METADATA_KEY)
1451 .map(|s| sanitise_avro_name(s))
1452 .unwrap_or_else(|| name_gen.make_unique(field_name));
1453 let mut obj = JsonMap::from_iter([
1454 ("type".into(), json!("fixed")),
1455 ("name".into(), json!(chosen_name)),
1456 ("size".into(), json!(len)),
1457 ]);
1458 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1459 obj.insert("namespace".into(), json!(ns));
1460 }
1461 Value::Object(obj)
1462 }
1463 }
1464 #[cfg(feature = "small_decimals")]
1465 DataType::Decimal32(precision, scale) | DataType::Decimal64(precision, scale) => {
1466 handle_decimal(precision, scale)?
1467 }
1468 DataType::Decimal128(precision, scale) | DataType::Decimal256(precision, scale) => {
1469 handle_decimal(precision, scale)?
1470 }
1471 DataType::Date32 => json!({ "type": "int", "logicalType": "date" }),
1472 #[cfg(not(feature = "avro_custom_types"))]
1473 DataType::Date64 => json!({ "type": "long", "logicalType": "local-timestamp-millis" }),
1474 #[cfg(feature = "avro_custom_types")]
1475 DataType::Date64 => json!({ "type": "long", "logicalType": "arrow.date64" }),
1476 DataType::Time32(unit) => match unit {
1477 TimeUnit::Millisecond => json!({ "type": "int", "logicalType": "time-millis" }),
1478 #[cfg(not(feature = "avro_custom_types"))]
1479 TimeUnit::Second => {
1480 if !strip {
1482 extras.insert("arrowTimeUnit".into(), Value::String("second".into()));
1483 }
1484 json!({ "type": "int", "logicalType": "time-millis" })
1485 }
1486 #[cfg(feature = "avro_custom_types")]
1487 TimeUnit::Second => {
1488 json!({ "type": "int", "logicalType": "arrow.time32-second" })
1489 }
1490 _ => Value::String("int".into()),
1491 },
1492 DataType::Time64(unit) => match unit {
1493 TimeUnit::Microsecond => json!({ "type": "long", "logicalType": "time-micros" }),
1494 #[cfg(not(feature = "avro_custom_types"))]
1495 TimeUnit::Nanosecond => {
1496 if !strip {
1498 extras.insert("arrowTimeUnit".into(), Value::String("nanosecond".into()));
1499 }
1500 json!({ "type": "long", "logicalType": "time-micros" })
1501 }
1502 #[cfg(feature = "avro_custom_types")]
1503 TimeUnit::Nanosecond => {
1504 json!({ "type": "long", "logicalType": "arrow.time64-nanosecond" })
1505 }
1506 _ => Value::String("long".into()),
1507 },
1508 DataType::Timestamp(unit, tz) => {
1509 #[cfg(feature = "avro_custom_types")]
1510 if matches!(unit, TimeUnit::Second) {
1511 let logical_type = if tz.is_some() {
1512 "arrow.timestamp-second"
1513 } else {
1514 "arrow.local-timestamp-second"
1515 };
1516 return Ok((
1517 json!({ "type": "long", "logicalType": logical_type }),
1518 extras,
1519 ));
1520 }
1521 let logical_type = match (unit, tz.is_some()) {
1522 (TimeUnit::Millisecond, true) => "timestamp-millis",
1523 (TimeUnit::Millisecond, false) => "local-timestamp-millis",
1524 (TimeUnit::Microsecond, true) => "timestamp-micros",
1525 (TimeUnit::Microsecond, false) => "local-timestamp-micros",
1526 (TimeUnit::Nanosecond, true) => "timestamp-nanos",
1527 (TimeUnit::Nanosecond, false) => "local-timestamp-nanos",
1528 (TimeUnit::Second, has_tz) => {
1529 if !strip {
1531 extras.insert("arrowTimeUnit".into(), Value::String("second".into()));
1532 }
1533 let ts_logical_type = if has_tz {
1534 "timestamp-millis"
1535 } else {
1536 "local-timestamp-millis"
1537 };
1538 return Ok((
1539 json!({ "type": "long", "logicalType": ts_logical_type }),
1540 extras,
1541 ));
1542 }
1543 };
1544 if !strip && matches!(unit, TimeUnit::Nanosecond) {
1545 extras.insert("arrowTimeUnit".into(), Value::String("nanosecond".into()));
1546 }
1547 json!({ "type": "long", "logicalType": logical_type })
1548 }
1549 #[cfg(not(feature = "avro_custom_types"))]
1550 DataType::Duration(_unit) => Value::String("long".into()),
1551 #[cfg(feature = "avro_custom_types")]
1552 DataType::Duration(unit) => {
1553 let logical_type = match unit {
1556 TimeUnit::Second => "arrow.duration-seconds",
1557 TimeUnit::Millisecond => "arrow.duration-millis",
1558 TimeUnit::Microsecond => "arrow.duration-micros",
1559 TimeUnit::Nanosecond => "arrow.duration-nanos",
1560 };
1561 json!({ "type": "long", "logicalType": logical_type })
1562 }
1563 #[cfg(not(feature = "avro_custom_types"))]
1564 DataType::Interval(IntervalUnit::MonthDayNano) => {
1565 let chosen_name = metadata
1567 .get(AVRO_NAME_METADATA_KEY)
1568 .map(|s| sanitise_avro_name(s))
1569 .unwrap_or_else(|| name_gen.make_unique(field_name));
1570 let mut obj = JsonMap::from_iter([
1571 ("type".into(), json!("fixed")),
1572 ("name".into(), json!(chosen_name)),
1573 ("size".into(), json!(12)),
1574 ("logicalType".into(), json!("duration")),
1575 ]);
1576 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1577 obj.insert("namespace".into(), json!(ns));
1578 }
1579 json!(obj)
1580 }
1581 #[cfg(feature = "avro_custom_types")]
1582 DataType::Interval(IntervalUnit::MonthDayNano) => {
1583 let chosen_name = metadata
1586 .get(AVRO_NAME_METADATA_KEY)
1587 .map(|s| sanitise_avro_name(s))
1588 .unwrap_or_else(|| name_gen.make_unique(field_name));
1589 let mut obj = JsonMap::from_iter([
1590 ("type".into(), json!("fixed")),
1591 ("name".into(), json!(chosen_name)),
1592 ("size".into(), json!(16)),
1593 ("logicalType".into(), json!("arrow.interval-month-day-nano")),
1594 ]);
1595 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1596 obj.insert("namespace".into(), json!(ns));
1597 }
1598 json!(obj)
1599 }
1600 #[cfg(not(feature = "avro_custom_types"))]
1601 DataType::Interval(IntervalUnit::YearMonth) => {
1602 let chosen_name = metadata
1604 .get(AVRO_NAME_METADATA_KEY)
1605 .map(|s| sanitise_avro_name(s))
1606 .unwrap_or_else(|| name_gen.make_unique(field_name));
1607 let mut extras = JsonMap::from_iter([
1608 ("type".into(), json!("fixed")),
1609 ("name".into(), json!(chosen_name)),
1610 ("size".into(), json!(12)),
1611 ("logicalType".into(), json!("duration")),
1612 ]);
1613 if !strip {
1614 extras.insert(
1615 "arrowIntervalUnit".into(),
1616 Value::String("yearmonth".into()),
1617 );
1618 }
1619 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1620 extras.insert("namespace".into(), json!(ns));
1621 }
1622 json!(extras)
1623 }
1624 #[cfg(feature = "avro_custom_types")]
1625 DataType::Interval(IntervalUnit::YearMonth) => {
1626 let chosen_name = metadata
1627 .get(AVRO_NAME_METADATA_KEY)
1628 .map(|s| sanitise_avro_name(s))
1629 .unwrap_or_else(|| name_gen.make_unique(field_name));
1630 let mut obj = JsonMap::from_iter([
1631 ("type".into(), json!("fixed")),
1632 ("name".into(), json!(chosen_name)),
1633 ("size".into(), json!(4)),
1634 ("logicalType".into(), json!("arrow.interval-year-month")),
1635 ]);
1636 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1637 obj.insert("namespace".into(), json!(ns));
1638 }
1639 json!(obj)
1640 }
1641 #[cfg(not(feature = "avro_custom_types"))]
1642 DataType::Interval(IntervalUnit::DayTime) => {
1643 let chosen_name = metadata
1645 .get(AVRO_NAME_METADATA_KEY)
1646 .map(|s| sanitise_avro_name(s))
1647 .unwrap_or_else(|| name_gen.make_unique(field_name));
1648 let mut obj = JsonMap::from_iter([
1649 ("type".into(), json!("fixed")),
1650 ("name".into(), json!(chosen_name)),
1651 ("size".into(), json!(12)),
1652 ("logicalType".into(), json!("duration")),
1653 ]);
1654 if !strip {
1655 obj.insert("arrowIntervalUnit".into(), Value::String("daytime".into()));
1656 }
1657 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1658 obj.insert("namespace".into(), json!(ns));
1659 }
1660 json!(obj)
1661 }
1662 #[cfg(feature = "avro_custom_types")]
1663 DataType::Interval(IntervalUnit::DayTime) => {
1664 let chosen_name = metadata
1665 .get(AVRO_NAME_METADATA_KEY)
1666 .map(|s| sanitise_avro_name(s))
1667 .unwrap_or_else(|| name_gen.make_unique(field_name));
1668 let mut obj = JsonMap::from_iter([
1669 ("type".into(), json!("fixed")),
1670 ("name".into(), json!(chosen_name)),
1671 ("size".into(), json!(8)),
1672 ("logicalType".into(), json!("arrow.interval-day-time")),
1673 ]);
1674 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1675 obj.insert("namespace".into(), json!(ns));
1676 }
1677 json!(obj)
1678 }
1679 DataType::List(child) | DataType::LargeList(child) => {
1680 if matches!(dt, DataType::LargeList(_)) && !strip {
1681 extras.insert("arrowLargeList".into(), Value::Bool(true));
1682 }
1683 let items_schema = process_datatype(
1684 child.data_type(),
1685 child.name(),
1686 child.metadata(),
1687 name_gen,
1688 null_order,
1689 child.is_nullable(),
1690 strip,
1691 )?;
1692 json!({
1693 "type": "array",
1694 "items": items_schema
1695 })
1696 }
1697 DataType::ListView(child) | DataType::LargeListView(child) => {
1698 if matches!(dt, DataType::LargeListView(_)) && !strip {
1699 extras.insert("arrowLargeList".into(), Value::Bool(true));
1700 }
1701 if !strip {
1702 extras.insert("arrowListView".into(), Value::Bool(true));
1703 }
1704 let items_schema = process_datatype(
1705 child.data_type(),
1706 child.name(),
1707 child.metadata(),
1708 name_gen,
1709 null_order,
1710 child.is_nullable(),
1711 strip,
1712 )?;
1713 json!({
1714 "type": "array",
1715 "items": items_schema
1716 })
1717 }
1718 DataType::FixedSizeList(child, len) => {
1719 if !strip {
1720 extras.insert("arrowFixedSize".into(), json!(len));
1721 }
1722 let items_schema = process_datatype(
1723 child.data_type(),
1724 child.name(),
1725 child.metadata(),
1726 name_gen,
1727 null_order,
1728 child.is_nullable(),
1729 strip,
1730 )?;
1731 json!({
1732 "type": "array",
1733 "items": items_schema
1734 })
1735 }
1736 DataType::Map(entries, _) => {
1737 let value_field = match entries.data_type() {
1738 DataType::Struct(fs) => &fs[1],
1739 _ => {
1740 return Err(ArrowError::SchemaError(
1741 "Map 'entries' field must be Struct(key,value)".into(),
1742 ));
1743 }
1744 };
1745 let values_schema = process_datatype(
1746 value_field.data_type(),
1747 value_field.name(),
1748 value_field.metadata(),
1749 name_gen,
1750 null_order,
1751 value_field.is_nullable(),
1752 strip,
1753 )?;
1754 json!({
1755 "type": "map",
1756 "values": values_schema
1757 })
1758 }
1759 DataType::Struct(fields) => {
1760 let avro_fields = fields
1761 .iter()
1762 .map(|field| arrow_field_to_avro(field, name_gen, null_order, strip))
1763 .collect::<Result<Vec<_>, _>>()?;
1764 let chosen_name = metadata
1766 .get(AVRO_NAME_METADATA_KEY)
1767 .map(|s| sanitise_avro_name(s))
1768 .unwrap_or_else(|| name_gen.make_unique(field_name));
1769 let mut obj = JsonMap::from_iter([
1770 ("type".into(), json!("record")),
1771 ("name".into(), json!(chosen_name)),
1772 ("fields".into(), Value::Array(avro_fields)),
1773 ]);
1774 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1775 obj.insert("namespace".into(), json!(ns));
1776 }
1777 Value::Object(obj)
1778 }
1779 DataType::Dictionary(_, value) => {
1780 if let Some(j) = metadata.get(AVRO_ENUM_SYMBOLS_METADATA_KEY) {
1781 let symbols: Vec<&str> =
1782 serde_json::from_str(j).map_err(|e| ArrowError::ParseError(e.to_string()))?;
1783 let chosen_name = metadata
1785 .get(AVRO_NAME_METADATA_KEY)
1786 .map(|s| sanitise_avro_name(s))
1787 .unwrap_or_else(|| name_gen.make_unique(field_name));
1788 let mut obj = JsonMap::from_iter([
1789 ("type".into(), json!("enum")),
1790 ("name".into(), json!(chosen_name)),
1791 ("symbols".into(), json!(symbols)),
1792 ]);
1793 if let Some(ns) = metadata.get(AVRO_NAMESPACE_METADATA_KEY) {
1794 obj.insert("namespace".into(), json!(ns));
1795 }
1796 Value::Object(obj)
1797 } else {
1798 process_datatype(
1799 value.as_ref(),
1800 field_name,
1801 metadata,
1802 name_gen,
1803 null_order,
1804 false,
1805 strip,
1806 )?
1807 }
1808 }
1809 #[cfg(feature = "avro_custom_types")]
1810 DataType::RunEndEncoded(run_ends, values) => {
1811 let bits = match run_ends.data_type() {
1812 DataType::Int16 => 16,
1813 DataType::Int32 => 32,
1814 DataType::Int64 => 64,
1815 other => {
1816 return Err(ArrowError::SchemaError(format!(
1817 "RunEndEncoded requires Int16/Int32/Int64 for run_ends, found: {other:?}"
1818 )));
1819 }
1820 };
1821 let (value_schema, value_extras) = datatype_to_avro(
1823 values.data_type(),
1824 values.name(),
1825 values.metadata(),
1826 name_gen,
1827 null_order,
1828 strip,
1829 )?;
1830 let mut merged = merge_extras(value_schema, value_extras);
1831 if values.is_nullable() {
1832 merged = wrap_nullable(merged, null_order);
1833 }
1834 let mut extras = JsonMap::new();
1835 extras.insert("logicalType".into(), json!("arrow.run-end-encoded"));
1836 extras.insert("arrow.runEndIndexBits".into(), json!(bits));
1837 return Ok((merged, extras));
1838 }
1839 #[cfg(not(feature = "avro_custom_types"))]
1840 DataType::RunEndEncoded(_run_ends, values) => {
1841 let (value_schema, _extras) = datatype_to_avro(
1842 values.data_type(),
1843 values.name(),
1844 values.metadata(),
1845 name_gen,
1846 null_order,
1847 strip,
1848 )?;
1849 return Ok((value_schema, JsonMap::new()));
1850 }
1851 DataType::Union(fields, mode) => {
1852 let mut branches: Vec<Value> = Vec::with_capacity(fields.len());
1853 let mut type_ids: Vec<i32> = Vec::with_capacity(fields.len());
1854 for (type_id, field_ref) in fields.iter() {
1855 let (branch_schema, _branch_extras) = datatype_to_avro(
1857 field_ref.data_type(),
1858 field_ref.name(),
1859 field_ref.metadata(),
1860 name_gen,
1861 null_order,
1862 strip,
1863 )?;
1864 if matches!(branch_schema, Value::Array(_)) {
1866 return Err(ArrowError::SchemaError(
1867 "Avro union may not immediately contain another union".into(),
1868 ));
1869 }
1870 branches.push(branch_schema);
1871 type_ids.push(type_id as i32);
1872 }
1873 let mut seen: HashSet<String> = HashSet::with_capacity(branches.len());
1874 for b in &branches {
1875 let sig = union_branch_signature(b)?;
1876 if !seen.insert(sig) {
1877 return Err(ArrowError::SchemaError(
1878 "Avro union contains duplicate branch types (disallowed by spec)".into(),
1879 ));
1880 }
1881 }
1882 if !strip {
1883 extras.insert(
1884 "arrowUnionMode".into(),
1885 Value::String(
1886 match mode {
1887 UnionMode::Sparse => "sparse",
1888 UnionMode::Dense => "dense",
1889 }
1890 .to_string(),
1891 ),
1892 );
1893 extras.insert(
1894 "arrowUnionTypeIds".into(),
1895 Value::Array(type_ids.into_iter().map(|id| json!(id)).collect()),
1896 );
1897 }
1898 Value::Array(branches)
1899 }
1900 #[cfg(not(feature = "small_decimals"))]
1901 other => {
1902 return Err(ArrowError::NotYetImplemented(format!(
1903 "Arrow type {other:?} has no Avro representation"
1904 )));
1905 }
1906 };
1907 Ok((val, extras))
1908}
1909
1910fn process_datatype(
1911 dt: &DataType,
1912 field_name: &str,
1913 metadata: &Metadata,
1914 name_gen: &mut NameGenerator,
1915 null_order: Nullability,
1916 is_nullable: bool,
1917 strip: bool,
1918) -> Result<Value, ArrowError> {
1919 let (schema, extras) = datatype_to_avro(dt, field_name, metadata, name_gen, null_order, strip)?;
1920 let mut merged = merge_extras(schema, extras);
1921 if is_nullable {
1922 merged = wrap_nullable(merged, null_order)
1923 }
1924 Ok(merged)
1925}
1926
1927fn arrow_field_to_avro(
1928 field: &ArrowField,
1929 name_gen: &mut NameGenerator,
1930 null_order: Nullability,
1931 strip: bool,
1932) -> Result<Value, ArrowError> {
1933 let avro_name = sanitise_avro_name(field.name());
1934 let schema_value = process_datatype(
1935 field.data_type(),
1936 &avro_name,
1937 field.metadata(),
1938 name_gen,
1939 null_order,
1940 field.is_nullable(),
1941 strip,
1942 )?;
1943 let mut map = JsonMap::with_capacity(field.metadata().len() + 3);
1945 map.insert("name".into(), Value::String(avro_name));
1946 map.insert("type".into(), schema_value);
1947 for (meta_key, meta_val) in field.metadata() {
1949 if is_internal_arrow_key(meta_key) {
1950 continue;
1951 }
1952 match meta_key.as_str() {
1953 AVRO_DOC_METADATA_KEY => {
1954 map.insert("doc".into(), Value::String(meta_val.clone()));
1955 }
1956 AVRO_FIELD_DEFAULT_METADATA_KEY => {
1957 let default_value = serde_json::from_str(meta_val)
1958 .unwrap_or_else(|_| Value::String(meta_val.clone()));
1959 map.insert("default".into(), default_value);
1960 }
1961 _ => {
1962 let json_val = serde_json::from_str(meta_val)
1963 .unwrap_or_else(|_| Value::String(meta_val.clone()));
1964 map.insert(meta_key.clone(), json_val);
1965 }
1966 }
1967 }
1968 Ok(Value::Object(map))
1969}
1970
1971#[cfg(test)]
1972mod tests {
1973 use super::*;
1974 use crate::codec::{AvroField, AvroFieldBuilder};
1975 use arrow_schema::{DataType, Fields, SchemaBuilder, TimeUnit, UnionFields};
1976 use serde_json::json;
1977 use std::sync::Arc;
1978
1979 fn int_schema() -> Schema<'static> {
1980 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))
1981 }
1982
1983 fn record_schema() -> Schema<'static> {
1984 Schema::Complex(ComplexType::Record(Record {
1985 name: "record1",
1986 namespace: Some("test.namespace"),
1987 doc: Some(Cow::from("A test record")),
1988 aliases: vec![],
1989 fields: vec![
1990 Field {
1991 name: "field1",
1992 doc: Some(Cow::from("An integer field")),
1993 r#type: int_schema(),
1994 default: None,
1995 aliases: vec![],
1996 },
1997 Field {
1998 name: "field2",
1999 doc: None,
2000 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2001 default: None,
2002 aliases: vec![],
2003 },
2004 ],
2005 attributes: Attributes::default(),
2006 }))
2007 }
2008
2009 fn single_field_schema(field: ArrowField) -> arrow_schema::Schema {
2010 let mut sb = SchemaBuilder::new();
2011 sb.push(field);
2012 sb.finish()
2013 }
2014
2015 fn assert_json_contains(avro_json: &str, needle: &str) {
2016 assert!(
2017 avro_json.contains(needle),
2018 "JSON did not contain `{needle}` : {avro_json}"
2019 )
2020 }
2021
2022 #[test]
2023 fn test_deserialize() {
2024 let t: Schema = serde_json::from_str("\"string\"").unwrap();
2025 assert_eq!(
2026 t,
2027 Schema::TypeName(TypeName::Primitive(PrimitiveType::String))
2028 );
2029
2030 let t: Schema = serde_json::from_str("[\"int\", \"null\"]").unwrap();
2031 assert_eq!(
2032 t,
2033 Schema::Union(vec![
2034 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2035 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2036 ])
2037 );
2038
2039 let t: Type = serde_json::from_str(
2040 r#"{
2041 "type":"long",
2042 "logicalType":"timestamp-micros"
2043 }"#,
2044 )
2045 .unwrap();
2046
2047 let timestamp = Type {
2048 r#type: TypeName::Primitive(PrimitiveType::Long),
2049 attributes: Attributes {
2050 logical_type: Some("timestamp-micros"),
2051 additional: Default::default(),
2052 },
2053 };
2054
2055 assert_eq!(t, timestamp);
2056
2057 let t: ComplexType = serde_json::from_str(
2058 r#"{
2059 "type":"fixed",
2060 "name":"fixed",
2061 "namespace":"topLevelRecord.value",
2062 "size":11,
2063 "logicalType":"decimal",
2064 "precision":25,
2065 "scale":2
2066 }"#,
2067 )
2068 .unwrap();
2069
2070 let decimal = ComplexType::Fixed(Fixed {
2071 name: "fixed",
2072 namespace: Some("topLevelRecord.value"),
2073 aliases: vec![],
2074 size: 11,
2075 attributes: Attributes {
2076 logical_type: Some("decimal"),
2077 additional: vec![("precision", json!(25)), ("scale", json!(2))]
2078 .into_iter()
2079 .collect(),
2080 },
2081 });
2082
2083 assert_eq!(t, decimal);
2084
2085 let schema: Schema = serde_json::from_str(
2086 r#"{
2087 "type":"record",
2088 "name":"topLevelRecord",
2089 "fields":[
2090 {
2091 "name":"value",
2092 "type":[
2093 {
2094 "type":"fixed",
2095 "name":"fixed",
2096 "namespace":"topLevelRecord.value",
2097 "size":11,
2098 "logicalType":"decimal",
2099 "precision":25,
2100 "scale":2
2101 },
2102 "null"
2103 ]
2104 }
2105 ]
2106 }"#,
2107 )
2108 .unwrap();
2109
2110 assert_eq!(
2111 schema,
2112 Schema::Complex(ComplexType::Record(Record {
2113 name: "topLevelRecord",
2114 namespace: None,
2115 doc: None,
2116 aliases: vec![],
2117 fields: vec![Field {
2118 name: "value",
2119 doc: None,
2120 r#type: Schema::Union(vec![
2121 Schema::Complex(decimal),
2122 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2123 ]),
2124 default: None,
2125 aliases: vec![],
2126 },],
2127 attributes: Default::default(),
2128 }))
2129 );
2130
2131 let schema: Schema = serde_json::from_str(
2132 r#"{
2133 "type": "record",
2134 "name": "LongList",
2135 "aliases": ["LinkedLongs"],
2136 "fields" : [
2137 {"name": "value", "type": "long"},
2138 {"name": "next", "type": ["null", "LongList"]}
2139 ]
2140 }"#,
2141 )
2142 .unwrap();
2143
2144 assert_eq!(
2145 schema,
2146 Schema::Complex(ComplexType::Record(Record {
2147 name: "LongList",
2148 namespace: None,
2149 doc: None,
2150 aliases: vec!["LinkedLongs"],
2151 fields: vec![
2152 Field {
2153 name: "value",
2154 doc: None,
2155 r#type: Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
2156 default: None,
2157 aliases: vec![],
2158 },
2159 Field {
2160 name: "next",
2161 doc: None,
2162 r#type: Schema::Union(vec![
2163 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2164 Schema::TypeName(TypeName::Ref("LongList")),
2165 ]),
2166 default: None,
2167 aliases: vec![],
2168 }
2169 ],
2170 attributes: Attributes::default(),
2171 }))
2172 );
2173
2174 let err = AvroField::try_from(&schema).unwrap_err().to_string();
2176 assert_eq!(err, "Parser error: Failed to resolve .LongList");
2177
2178 let schema: Schema = serde_json::from_str(
2179 r#"{
2180 "type":"record",
2181 "name":"topLevelRecord",
2182 "fields":[
2183 {
2184 "name":"id",
2185 "type":[
2186 "int",
2187 "null"
2188 ]
2189 },
2190 {
2191 "name":"timestamp_col",
2192 "type":[
2193 {
2194 "type":"long",
2195 "logicalType":"timestamp-micros"
2196 },
2197 "null"
2198 ]
2199 }
2200 ]
2201 }"#,
2202 )
2203 .unwrap();
2204
2205 assert_eq!(
2206 schema,
2207 Schema::Complex(ComplexType::Record(Record {
2208 name: "topLevelRecord",
2209 namespace: None,
2210 doc: None,
2211 aliases: vec![],
2212 fields: vec![
2213 Field {
2214 name: "id",
2215 doc: None,
2216 r#type: Schema::Union(vec![
2217 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2218 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2219 ]),
2220 default: None,
2221 aliases: vec![],
2222 },
2223 Field {
2224 name: "timestamp_col",
2225 doc: None,
2226 r#type: Schema::Union(vec![
2227 Schema::Type(timestamp),
2228 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2229 ]),
2230 default: None,
2231 aliases: vec![],
2232 }
2233 ],
2234 attributes: Default::default(),
2235 }))
2236 );
2237 let codec = AvroField::try_from(&schema).unwrap();
2238 let expected_arrow_field = arrow_schema::Field::new(
2239 "topLevelRecord",
2240 DataType::Struct(Fields::from(vec![
2241 arrow_schema::Field::new("id", DataType::Int32, true),
2242 arrow_schema::Field::new(
2243 "timestamp_col",
2244 DataType::Timestamp(TimeUnit::Microsecond, Some("+00:00".into())),
2245 true,
2246 ),
2247 ])),
2248 false,
2249 )
2250 .with_metadata(std::collections::HashMap::from([(
2251 AVRO_NAME_METADATA_KEY.to_string(),
2252 "topLevelRecord".to_string(),
2253 )]));
2254
2255 assert_eq!(codec.field(), expected_arrow_field);
2256
2257 let schema: Schema = serde_json::from_str(
2258 r#"{
2259 "type": "record",
2260 "name": "HandshakeRequest", "namespace":"org.apache.avro.ipc",
2261 "fields": [
2262 {"name": "clientHash", "type": {"type": "fixed", "name": "MD5", "size": 16}},
2263 {"name": "clientProtocol", "type": ["null", "string"]},
2264 {"name": "serverHash", "type": "MD5"},
2265 {"name": "meta", "type": ["null", {"type": "map", "values": "bytes"}]}
2266 ]
2267 }"#,
2268 )
2269 .unwrap();
2270
2271 assert_eq!(
2272 schema,
2273 Schema::Complex(ComplexType::Record(Record {
2274 name: "HandshakeRequest",
2275 namespace: Some("org.apache.avro.ipc"),
2276 doc: None,
2277 aliases: vec![],
2278 fields: vec![
2279 Field {
2280 name: "clientHash",
2281 doc: None,
2282 r#type: Schema::Complex(ComplexType::Fixed(Fixed {
2283 name: "MD5",
2284 namespace: None,
2285 aliases: vec![],
2286 size: 16,
2287 attributes: Default::default(),
2288 })),
2289 default: None,
2290 aliases: vec![],
2291 },
2292 Field {
2293 name: "clientProtocol",
2294 doc: None,
2295 r#type: Schema::Union(vec![
2296 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2297 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2298 ]),
2299 default: None,
2300 aliases: vec![],
2301 },
2302 Field {
2303 name: "serverHash",
2304 doc: None,
2305 r#type: Schema::TypeName(TypeName::Ref("MD5")),
2306 default: None,
2307 aliases: vec![],
2308 },
2309 Field {
2310 name: "meta",
2311 doc: None,
2312 r#type: Schema::Union(vec![
2313 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
2314 Schema::Complex(ComplexType::Map(Map {
2315 values: Box::new(Schema::TypeName(TypeName::Primitive(
2316 PrimitiveType::Bytes
2317 ))),
2318 attributes: Default::default(),
2319 })),
2320 ]),
2321 default: None,
2322 aliases: vec![],
2323 }
2324 ],
2325 attributes: Default::default(),
2326 }))
2327 );
2328 }
2329
2330 #[test]
2331 fn test_canonical_form_generation_comprehensive_record() {
2332 let json_str = r#"{
2334 "type": "record",
2335 "name": "E2eComprehensive",
2336 "namespace": "org.apache.arrow.avrotests.v1",
2337 "doc": "Comprehensive Avro writer schema to exercise arrow-avro Reader/Decoder paths.",
2338 "fields": [
2339 {"name": "id", "type": "long", "doc": "Primary row id", "aliases": ["identifier"]},
2340 {"name": "flag", "type": "boolean", "default": true, "doc": "A sample boolean with default true"},
2341 {"name": "ratio_f32", "type": "float", "default": 0.0, "doc": "Float32 example"},
2342 {"name": "ratio_f64", "type": "double", "default": 0.0, "doc": "Float64 example"},
2343 {"name": "count_i32", "type": "int", "default": 0, "doc": "Int32 example"},
2344 {"name": "count_i64", "type": "long", "default": 0, "doc": "Int64 example"},
2345 {"name": "opt_i32_nullfirst", "type": ["null", "int"], "default": null, "doc": "Nullable int (null-first)"},
2346 {"name": "opt_str_nullsecond", "type": ["string", "null"], "default": "", "aliases": ["old_opt_str"], "doc": "Nullable string (null-second). Default is empty string."},
2347 {"name": "tri_union_prim", "type": ["int", "string", "boolean"], "default": 0, "doc": "Union[int, string, boolean] with default on first branch (int=0)."},
2348 {"name": "str_utf8", "type": "string", "default": "default", "doc": "Plain Utf8 string (Reader may use Utf8View)."},
2349 {"name": "raw_bytes", "type": "bytes", "default": "", "doc": "Raw bytes field"},
2350 {"name": "fx16_plain", "type": {"type": "fixed", "name": "Fx16", "namespace": "org.apache.arrow.avrotests.v1.types", "aliases": ["Fixed16Old"], "size": 16}, "doc": "Plain fixed(16)"},
2351 {"name": "dec_bytes_s10_2", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}, "doc": "Decimal encoded on bytes, precision 10, scale 2"},
2352 {"name": "dec_fix_s20_4", "type": {"type": "fixed", "name": "DecFix20", "namespace": "org.apache.arrow.avrotests.v1.types", "size": 20, "logicalType": "decimal", "precision": 20, "scale": 4}, "doc": "Decimal encoded on fixed(20), precision 20, scale 4"},
2353 {"name": "uuid_str", "type": {"type": "string", "logicalType": "uuid"}, "doc": "UUID logical type on string"},
2354 {"name": "d_date", "type": {"type": "int", "logicalType": "date"}, "doc": "Date32: days since 1970-01-01"},
2355 {"name": "t_millis", "type": {"type": "int", "logicalType": "time-millis"}, "doc": "Time32-millis"},
2356 {"name": "t_micros", "type": {"type": "long", "logicalType": "time-micros"}, "doc": "Time64-micros"},
2357 {"name": "ts_millis_utc", "type": {"type": "long", "logicalType": "timestamp-millis"}, "doc": "Timestamp ms (UTC)"},
2358 {"name": "ts_micros_utc", "type": {"type": "long", "logicalType": "timestamp-micros"}, "doc": "Timestamp µs (UTC)"},
2359 {"name": "ts_millis_local", "type": {"type": "long", "logicalType": "local-timestamp-millis"}, "doc": "Local timestamp ms"},
2360 {"name": "ts_micros_local", "type": {"type": "long", "logicalType": "local-timestamp-micros"}, "doc": "Local timestamp µs"},
2361 {"name": "interval_mdn", "type": {"type": "fixed", "name": "Dur12", "namespace": "org.apache.arrow.avrotests.v1.types", "size": 12, "logicalType": "duration"}, "doc": "Duration: fixed(12) little-endian (months, days, millis)"},
2362 {"name": "status", "type": {"type": "enum", "name": "Status", "namespace": "org.apache.arrow.avrotests.v1.types", "symbols": ["UNKNOWN", "NEW", "PROCESSING", "DONE"], "aliases": ["State"], "doc": "Processing status enum with default"}, "default": "UNKNOWN", "doc": "Enum field using default when resolving"},
2363 {"name": "arr_union", "type": {"type": "array", "items": ["long", "string", "null"]}, "default": [], "doc": "Array whose items are a union[long,string,null]"},
2364 {"name": "map_union", "type": {"type": "map", "values": ["null", "double", "string"]}, "default": {}, "doc": "Map whose values are a union[null,double,string]"},
2365 {"name": "address", "type": {"type": "record", "name": "Address", "namespace": "org.apache.arrow.avrotests.v1.types", "doc": "Postal address with defaults and field alias", "fields": [
2366 {"name": "street", "type": "string", "default": "", "aliases": ["street_name"], "doc": "Street (field alias = street_name)"},
2367 {"name": "zip", "type": "int", "default": 0, "doc": "ZIP/postal code"},
2368 {"name": "country", "type": "string", "default": "US", "doc": "Country code"}
2369 ]}, "doc": "Embedded Address record"},
2370 {"name": "maybe_auth", "type": {"type": "record", "name": "MaybeAuth", "namespace": "org.apache.arrow.avrotests.v1.types", "doc": "Optional auth token model", "fields": [
2371 {"name": "user", "type": "string", "doc": "Username"},
2372 {"name": "token", "type": ["null", "bytes"], "default": null, "doc": "Nullable auth token"}
2373 ]}},
2374 {"name": "union_enum_record_array_map", "type": [
2375 {"type": "enum", "name": "Color", "namespace": "org.apache.arrow.avrotests.v1.types", "symbols": ["RED", "GREEN", "BLUE"], "doc": "Color enum"},
2376 {"type": "record", "name": "RecA", "namespace": "org.apache.arrow.avrotests.v1.types", "fields": [{"name": "a", "type": "int"}, {"name": "b", "type": "string"}]},
2377 {"type": "record", "name": "RecB", "namespace": "org.apache.arrow.avrotests.v1.types", "fields": [{"name": "x", "type": "long"}, {"name": "y", "type": "bytes"}]},
2378 {"type": "array", "items": "long"},
2379 {"type": "map", "values": "string"}
2380 ], "doc": "Union of enum, two records, array, and map"},
2381 {"name": "union_date_or_fixed4", "type": [
2382 {"type": "int", "logicalType": "date"},
2383 {"type": "fixed", "name": "Fx4", "size": 4}
2384 ], "doc": "Union of date(int) or fixed(4)"},
2385 {"name": "union_interval_or_string", "type": [
2386 {"type": "fixed", "name": "Dur12U", "size": 12, "logicalType": "duration"},
2387 "string"
2388 ], "doc": "Union of duration(fixed12) or string"},
2389 {"name": "union_uuid_or_fixed10", "type": [
2390 {"type": "string", "logicalType": "uuid"},
2391 {"type": "fixed", "name": "Fx10", "size": 10}
2392 ], "doc": "Union of UUID string or fixed(10)"},
2393 {"name": "array_records_with_union", "type": {"type": "array", "items": {
2394 "type": "record", "name": "KV", "namespace": "org.apache.arrow.avrotests.v1.types",
2395 "fields": [
2396 {"name": "key", "type": "string"},
2397 {"name": "val", "type": ["null", "int", "long"], "default": null}
2398 ]
2399 }}, "doc": "Array<record{key, val: union[null,int,long]}>", "default": []},
2400 {"name": "union_map_or_array_int", "type": [
2401 {"type": "map", "values": "int"},
2402 {"type": "array", "items": "int"}
2403 ], "doc": "Union[map<string,int>, array<int>]"},
2404 {"name": "renamed_with_default", "type": "int", "default": 42, "aliases": ["old_count"], "doc": "Field with alias and default"},
2405 {"name": "person", "type": {"type": "record", "name": "PersonV2", "namespace": "com.example.v2", "aliases": ["com.example.Person"], "doc": "Person record with alias pointing to previous namespace/name", "fields": [
2406 {"name": "name", "type": "string"},
2407 {"name": "age", "type": "int", "default": 0}
2408 ]}, "doc": "Record using type alias for schema evolution tests"}
2409 ]
2410 }"#;
2411 let avro = AvroSchema::new(json_str.to_string());
2412 let parsed = avro.schema().expect("schema should deserialize");
2413 let expected_canonical_form = r#"{"name":"org.apache.arrow.avrotests.v1.E2eComprehensive","type":"record","fields":[{"name":"id","type":"long"},{"name":"flag","type":"boolean"},{"name":"ratio_f32","type":"float"},{"name":"ratio_f64","type":"double"},{"name":"count_i32","type":"int"},{"name":"count_i64","type":"long"},{"name":"opt_i32_nullfirst","type":["null","int"]},{"name":"opt_str_nullsecond","type":["string","null"]},{"name":"tri_union_prim","type":["int","string","boolean"]},{"name":"str_utf8","type":"string"},{"name":"raw_bytes","type":"bytes"},{"name":"fx16_plain","type":{"name":"org.apache.arrow.avrotests.v1.types.Fx16","type":"fixed","size":16}},{"name":"dec_bytes_s10_2","type":"bytes"},{"name":"dec_fix_s20_4","type":{"name":"org.apache.arrow.avrotests.v1.types.DecFix20","type":"fixed","size":20}},{"name":"uuid_str","type":"string"},{"name":"d_date","type":"int"},{"name":"t_millis","type":"int"},{"name":"t_micros","type":"long"},{"name":"ts_millis_utc","type":"long"},{"name":"ts_micros_utc","type":"long"},{"name":"ts_millis_local","type":"long"},{"name":"ts_micros_local","type":"long"},{"name":"interval_mdn","type":{"name":"org.apache.arrow.avrotests.v1.types.Dur12","type":"fixed","size":12}},{"name":"status","type":{"name":"org.apache.arrow.avrotests.v1.types.Status","type":"enum","symbols":["UNKNOWN","NEW","PROCESSING","DONE"]}},{"name":"arr_union","type":{"type":"array","items":["long","string","null"]}},{"name":"map_union","type":{"type":"map","values":["null","double","string"]}},{"name":"address","type":{"name":"org.apache.arrow.avrotests.v1.types.Address","type":"record","fields":[{"name":"street","type":"string"},{"name":"zip","type":"int"},{"name":"country","type":"string"}]}},{"name":"maybe_auth","type":{"name":"org.apache.arrow.avrotests.v1.types.MaybeAuth","type":"record","fields":[{"name":"user","type":"string"},{"name":"token","type":["null","bytes"]}]}},{"name":"union_enum_record_array_map","type":[{"name":"org.apache.arrow.avrotests.v1.types.Color","type":"enum","symbols":["RED","GREEN","BLUE"]},{"name":"org.apache.arrow.avrotests.v1.types.RecA","type":"record","fields":[{"name":"a","type":"int"},{"name":"b","type":"string"}]},{"name":"org.apache.arrow.avrotests.v1.types.RecB","type":"record","fields":[{"name":"x","type":"long"},{"name":"y","type":"bytes"}]},{"type":"array","items":"long"},{"type":"map","values":"string"}]},{"name":"union_date_or_fixed4","type":["int",{"name":"org.apache.arrow.avrotests.v1.Fx4","type":"fixed","size":4}]},{"name":"union_interval_or_string","type":[{"name":"org.apache.arrow.avrotests.v1.Dur12U","type":"fixed","size":12},"string"]},{"name":"union_uuid_or_fixed10","type":["string",{"name":"org.apache.arrow.avrotests.v1.Fx10","type":"fixed","size":10}]},{"name":"array_records_with_union","type":{"type":"array","items":{"name":"org.apache.arrow.avrotests.v1.types.KV","type":"record","fields":[{"name":"key","type":"string"},{"name":"val","type":["null","int","long"]}]}}},{"name":"union_map_or_array_int","type":[{"type":"map","values":"int"},{"type":"array","items":"int"}]},{"name":"renamed_with_default","type":"int"},{"name":"person","type":{"name":"com.example.v2.PersonV2","type":"record","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]}}]}"#;
2414 let canonical_form =
2415 AvroSchema::generate_canonical_form(&parsed).expect("canonical form should be built");
2416 assert_eq!(
2417 canonical_form, expected_canonical_form,
2418 "Canonical form must match Avro spec PCF exactly"
2419 );
2420 }
2421
2422 #[test]
2423 fn test_new_schema_store() {
2424 let store = SchemaStore::new();
2425 assert!(store.schemas.is_empty());
2426 }
2427
2428 #[test]
2429 fn test_try_from_schemas_rabin() {
2430 let int_avro_schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2431 let record_avro_schema = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
2432 let mut schemas: HashMap<Fingerprint, AvroSchema> = HashMap::new();
2433 schemas.insert(
2434 int_avro_schema
2435 .fingerprint(FingerprintAlgorithm::Rabin)
2436 .unwrap(),
2437 int_avro_schema.clone(),
2438 );
2439 schemas.insert(
2440 record_avro_schema
2441 .fingerprint(FingerprintAlgorithm::Rabin)
2442 .unwrap(),
2443 record_avro_schema.clone(),
2444 );
2445 let store = SchemaStore::try_from(schemas).unwrap();
2446 let int_fp = int_avro_schema
2447 .fingerprint(FingerprintAlgorithm::Rabin)
2448 .unwrap();
2449 assert_eq!(store.lookup(&int_fp).cloned(), Some(int_avro_schema));
2450 let rec_fp = record_avro_schema
2451 .fingerprint(FingerprintAlgorithm::Rabin)
2452 .unwrap();
2453 assert_eq!(store.lookup(&rec_fp).cloned(), Some(record_avro_schema));
2454 }
2455
2456 #[test]
2457 fn test_try_from_with_duplicates() {
2458 let int_avro_schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2459 let record_avro_schema = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
2460 let mut schemas: HashMap<Fingerprint, AvroSchema> = HashMap::new();
2461 schemas.insert(
2462 int_avro_schema
2463 .fingerprint(FingerprintAlgorithm::Rabin)
2464 .unwrap(),
2465 int_avro_schema.clone(),
2466 );
2467 schemas.insert(
2468 record_avro_schema
2469 .fingerprint(FingerprintAlgorithm::Rabin)
2470 .unwrap(),
2471 record_avro_schema.clone(),
2472 );
2473 schemas.insert(
2475 int_avro_schema
2476 .fingerprint(FingerprintAlgorithm::Rabin)
2477 .unwrap(),
2478 int_avro_schema.clone(),
2479 );
2480 let store = SchemaStore::try_from(schemas).unwrap();
2481 assert_eq!(store.schemas.len(), 2);
2482 let int_fp = int_avro_schema
2483 .fingerprint(FingerprintAlgorithm::Rabin)
2484 .unwrap();
2485 assert_eq!(store.lookup(&int_fp).cloned(), Some(int_avro_schema));
2486 }
2487
2488 #[test]
2489 fn test_register_and_lookup_rabin() {
2490 let mut store = SchemaStore::new();
2491 let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2492 let fp_enum = store.register(schema.clone()).unwrap();
2493 match fp_enum {
2494 Fingerprint::Rabin(fp_val) => {
2495 assert_eq!(
2496 store.lookup(&Fingerprint::Rabin(fp_val)).cloned(),
2497 Some(schema.clone())
2498 );
2499 assert!(
2500 store
2501 .lookup(&Fingerprint::Rabin(fp_val.wrapping_add(1)))
2502 .is_none()
2503 );
2504 }
2505 Fingerprint::Id(_id) => {
2506 unreachable!("This test should only generate Rabin fingerprints")
2507 }
2508 Fingerprint::Id64(_id) => {
2509 unreachable!("This test should only generate Rabin fingerprints")
2510 }
2511 #[cfg(feature = "md5")]
2512 Fingerprint::MD5(_id) => {
2513 unreachable!("This test should only generate Rabin fingerprints")
2514 }
2515 #[cfg(feature = "sha256")]
2516 Fingerprint::SHA256(_id) => {
2517 unreachable!("This test should only generate Rabin fingerprints")
2518 }
2519 }
2520 }
2521
2522 #[test]
2523 fn test_set_and_lookup_id() {
2524 let mut store = SchemaStore::new();
2525 let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2526 let id = 42u32;
2527 let fp = Fingerprint::Id(id);
2528 let out_fp = store.set(fp, schema.clone()).unwrap();
2529 assert_eq!(out_fp, fp);
2530 assert_eq!(store.lookup(&fp).cloned(), Some(schema.clone()));
2531 assert!(store.lookup(&Fingerprint::Id(id.wrapping_add(1))).is_none());
2532 }
2533
2534 #[test]
2535 fn test_set_and_lookup_id64() {
2536 let mut store = SchemaStore::new();
2537 let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2538 let id64: u64 = 0xDEAD_BEEF_DEAD_BEEF;
2539 let fp = Fingerprint::Id64(id64);
2540 let out_fp = store.set(fp, schema.clone()).unwrap();
2541 assert_eq!(out_fp, fp, "set should return the same Id64 fingerprint");
2542 assert_eq!(
2543 store.lookup(&fp).cloned(),
2544 Some(schema.clone()),
2545 "lookup should find the schema by Id64"
2546 );
2547 assert!(
2548 store
2549 .lookup(&Fingerprint::Id64(id64.wrapping_add(1)))
2550 .is_none(),
2551 "lookup with a different Id64 must return None"
2552 );
2553 }
2554
2555 #[test]
2556 fn test_fingerprint_id64_conversions() {
2557 let algo_from_fp = FingerprintAlgorithm::from(&Fingerprint::Id64(123));
2558 assert_eq!(algo_from_fp, FingerprintAlgorithm::Id64);
2559 let fp_from_algo = Fingerprint::from(FingerprintAlgorithm::Id64);
2560 assert!(matches!(fp_from_algo, Fingerprint::Id64(0)));
2561 let strategy_from_fp = FingerprintStrategy::from(Fingerprint::Id64(5));
2562 assert!(matches!(strategy_from_fp, FingerprintStrategy::Id64(0)));
2563 let algo_from_strategy = FingerprintAlgorithm::from(strategy_from_fp);
2564 assert_eq!(algo_from_strategy, FingerprintAlgorithm::Id64);
2565 }
2566
2567 #[test]
2568 fn test_register_duplicate_schema() {
2569 let mut store = SchemaStore::new();
2570 let schema1 = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2571 let schema2 = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2572 let fingerprint1 = store.register(schema1).unwrap();
2573 let fingerprint2 = store.register(schema2).unwrap();
2574 assert_eq!(fingerprint1, fingerprint2);
2575 assert_eq!(store.schemas.len(), 1);
2576 }
2577
2578 #[test]
2579 fn test_set_and_lookup_with_provided_fingerprint() {
2580 let mut store = SchemaStore::new();
2581 let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2582 let fp = schema.fingerprint(FingerprintAlgorithm::Rabin).unwrap();
2583 let out_fp = store.set(fp, schema.clone()).unwrap();
2584 assert_eq!(out_fp, fp);
2585 assert_eq!(store.lookup(&fp).cloned(), Some(schema));
2586 }
2587
2588 #[test]
2589 fn test_set_duplicate_same_schema_ok() {
2590 let mut store = SchemaStore::new();
2591 let schema = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2592 let fp = schema.fingerprint(FingerprintAlgorithm::Rabin).unwrap();
2593 let _ = store.set(fp, schema.clone()).unwrap();
2594 let _ = store.set(fp, schema.clone()).unwrap();
2595 assert_eq!(store.schemas.len(), 1);
2596 }
2597
2598 #[test]
2599 fn test_set_duplicate_different_schema_collision_error() {
2600 let mut store = SchemaStore::new();
2601 let schema1 = AvroSchema::new(serde_json::to_string(&int_schema()).unwrap());
2602 let schema2 = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
2603 let fp = Fingerprint::Id(123);
2605 let _ = store.set(fp, schema1).unwrap();
2606 let err = store.set(fp, schema2).unwrap_err();
2607 let msg = format!("{err}");
2608 assert!(msg.contains("Schema fingerprint collision"));
2609 }
2610
2611 #[test]
2612 fn test_canonical_form_generation_primitive() {
2613 let schema = int_schema();
2614 let canonical_form = AvroSchema::generate_canonical_form(&schema).unwrap();
2615 assert_eq!(canonical_form, r#""int""#);
2616 }
2617
2618 #[test]
2619 fn test_canonical_form_generation_record() {
2620 let schema = record_schema();
2621 let expected_canonical_form = r#"{"name":"test.namespace.record1","type":"record","fields":[{"name":"field1","type":"int"},{"name":"field2","type":"string"}]}"#;
2622 let canonical_form = AvroSchema::generate_canonical_form(&schema).unwrap();
2623 assert_eq!(canonical_form, expected_canonical_form);
2624 }
2625
2626 #[test]
2627 fn test_fingerprint_calculation() {
2628 let canonical_form = r#"{"fields":[{"name":"a","type":"long"},{"name":"b","type":"string"}],"name":"test","type":"record"}"#;
2629 let expected_fingerprint = 10505236152925314060;
2630 let fingerprint = compute_fingerprint_rabin(canonical_form);
2631 assert_eq!(fingerprint, expected_fingerprint);
2632 }
2633
2634 #[test]
2635 fn test_register_and_lookup_complex_schema() {
2636 let mut store = SchemaStore::new();
2637 let schema = AvroSchema::new(serde_json::to_string(&record_schema()).unwrap());
2638 let canonical_form = r#"{"name":"test.namespace.record1","type":"record","fields":[{"name":"field1","type":"int"},{"name":"field2","type":"string"}]}"#;
2639 let expected_fingerprint = Fingerprint::Rabin(compute_fingerprint_rabin(canonical_form));
2640 let fingerprint = store.register(schema.clone()).unwrap();
2641 assert_eq!(fingerprint, expected_fingerprint);
2642 let looked_up = store.lookup(&fingerprint).cloned();
2643 assert_eq!(looked_up, Some(schema));
2644 }
2645
2646 #[test]
2647 fn test_fingerprints_returns_all_keys() {
2648 let mut store = SchemaStore::new();
2649 let fp_int = store
2650 .register(AvroSchema::new(
2651 serde_json::to_string(&int_schema()).unwrap(),
2652 ))
2653 .unwrap();
2654 let fp_record = store
2655 .register(AvroSchema::new(
2656 serde_json::to_string(&record_schema()).unwrap(),
2657 ))
2658 .unwrap();
2659 let fps = store.fingerprints();
2660 assert_eq!(fps.len(), 2);
2661 assert!(fps.contains(&fp_int));
2662 assert!(fps.contains(&fp_record));
2663 }
2664
2665 #[test]
2666 fn test_canonical_form_strips_attributes() {
2667 let schema_with_attrs = Schema::Complex(ComplexType::Record(Record {
2668 name: "record_with_attrs",
2669 namespace: None,
2670 doc: Some(Cow::from("This doc should be stripped")),
2671 aliases: vec!["alias1", "alias2"],
2672 fields: vec![Field {
2673 name: "f1",
2674 doc: Some(Cow::from("field doc")),
2675 r#type: Schema::Type(Type {
2676 r#type: TypeName::Primitive(PrimitiveType::Bytes),
2677 attributes: Attributes {
2678 logical_type: None,
2679 additional: HashMap::from([("precision", json!(4))]),
2680 },
2681 }),
2682 default: None,
2683 aliases: vec![],
2684 }],
2685 attributes: Attributes {
2686 logical_type: None,
2687 additional: HashMap::from([("custom_attr", json!("value"))]),
2688 },
2689 }));
2690 let expected_canonical_form = r#"{"name":"record_with_attrs","type":"record","fields":[{"name":"f1","type":"bytes"}]}"#;
2691 let canonical_form = AvroSchema::generate_canonical_form(&schema_with_attrs).unwrap();
2692 assert_eq!(canonical_form, expected_canonical_form);
2693 }
2694
2695 #[cfg(not(feature = "avro_custom_types"))]
2696 #[test]
2697 fn test_primitive_mappings() {
2698 let cases = vec![
2699 (DataType::Boolean, "\"boolean\""),
2700 (DataType::Int8, "\"int\""),
2701 (DataType::Int16, "\"int\""),
2702 (DataType::Int32, "\"int\""),
2703 (DataType::Int64, "\"long\""),
2704 (DataType::UInt8, "\"int\""),
2705 (DataType::UInt16, "\"int\""),
2706 (DataType::UInt32, "\"long\""),
2707 (DataType::UInt64, "\"long\""),
2708 (DataType::Float16, "\"float\""),
2709 (DataType::Float32, "\"float\""),
2710 (DataType::Float64, "\"double\""),
2711 (DataType::Utf8, "\"string\""),
2712 (DataType::Binary, "\"bytes\""),
2713 ];
2714 for (dt, avro_token) in cases {
2715 let field = ArrowField::new("col", dt.clone(), false);
2716 let arrow_schema = single_field_schema(field);
2717 let avro = AvroSchema::try_from(&arrow_schema).unwrap();
2718 assert_json_contains(&avro.json_string, avro_token);
2719 }
2720 }
2721
2722 #[cfg(feature = "avro_custom_types")]
2723 #[test]
2724 fn test_primitive_mappings() {
2725 let cases = vec![
2726 (DataType::Boolean, "\"boolean\""),
2727 (DataType::Int8, "\"logicalType\":\"arrow.int8\""),
2728 (DataType::Int16, "\"logicalType\":\"arrow.int16\""),
2729 (DataType::Int32, "\"int\""),
2730 (DataType::Int64, "\"long\""),
2731 (DataType::UInt8, "\"logicalType\":\"arrow.uint8\""),
2732 (DataType::UInt16, "\"logicalType\":\"arrow.uint16\""),
2733 (DataType::UInt32, "\"logicalType\":\"arrow.uint32\""),
2734 (DataType::UInt64, "\"logicalType\":\"arrow.uint64\""),
2735 (DataType::Float16, "\"logicalType\":\"arrow.float16\""),
2736 (DataType::Float32, "\"float\""),
2737 (DataType::Float64, "\"double\""),
2738 (DataType::Utf8, "\"string\""),
2739 (DataType::Binary, "\"bytes\""),
2740 ];
2741 for (dt, avro_token) in cases {
2742 let field = ArrowField::new("col", dt.clone(), false);
2743 let arrow_schema = single_field_schema(field);
2744 let avro = AvroSchema::try_from(&arrow_schema).unwrap();
2745 assert_json_contains(&avro.json_string, avro_token);
2746 }
2747 }
2748
2749 #[cfg(feature = "avro_custom_types")]
2750 #[test]
2751 fn test_custom_fixed_logical_types_preserve_namespace_metadata() {
2752 let namespace = "com.example.types";
2753
2754 let mut md_u64 = HashMap::new();
2755 md_u64.insert(AVRO_NAME_METADATA_KEY.to_string(), "U64Type".to_string());
2756 md_u64.insert(
2757 AVRO_NAMESPACE_METADATA_KEY.to_string(),
2758 namespace.to_string(),
2759 );
2760
2761 let mut md_f16 = HashMap::new();
2762 md_f16.insert(AVRO_NAME_METADATA_KEY.to_string(), "F16Type".to_string());
2763 md_f16.insert(
2764 AVRO_NAMESPACE_METADATA_KEY.to_string(),
2765 namespace.to_string(),
2766 );
2767
2768 let mut md_iv_ym = HashMap::new();
2769 md_iv_ym.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvYmType".to_string());
2770 md_iv_ym.insert(
2771 AVRO_NAMESPACE_METADATA_KEY.to_string(),
2772 namespace.to_string(),
2773 );
2774
2775 let mut md_iv_dt = HashMap::new();
2776 md_iv_dt.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvDtType".to_string());
2777 md_iv_dt.insert(
2778 AVRO_NAMESPACE_METADATA_KEY.to_string(),
2779 namespace.to_string(),
2780 );
2781
2782 let arrow_schema = ArrowSchema::new(vec![
2783 ArrowField::new("u64_col", DataType::UInt64, false).with_metadata(md_u64),
2784 ArrowField::new("f16_col", DataType::Float16, false).with_metadata(md_f16),
2785 ArrowField::new(
2786 "iv_ym_col",
2787 DataType::Interval(IntervalUnit::YearMonth),
2788 false,
2789 )
2790 .with_metadata(md_iv_ym),
2791 ArrowField::new(
2792 "iv_dt_col",
2793 DataType::Interval(IntervalUnit::DayTime),
2794 false,
2795 )
2796 .with_metadata(md_iv_dt),
2797 ]);
2798
2799 let avro = AvroSchema::try_from(&arrow_schema).unwrap();
2800 let root: Value = serde_json::from_str(&avro.json_string).unwrap();
2801 let fields = root
2802 .get("fields")
2803 .and_then(|f| f.as_array())
2804 .expect("record fields array");
2805
2806 let expected = [
2807 ("u64_col", "arrow.uint64"),
2808 ("f16_col", "arrow.float16"),
2809 ("iv_ym_col", "arrow.interval-year-month"),
2810 ("iv_dt_col", "arrow.interval-day-time"),
2811 ];
2812
2813 for (field_name, logical_type) in expected {
2814 let field = fields
2815 .iter()
2816 .find(|f| f.get("name").and_then(Value::as_str) == Some(field_name))
2817 .unwrap_or_else(|| panic!("missing field {field_name}"));
2818 let ty = field
2819 .get("type")
2820 .and_then(Value::as_object)
2821 .unwrap_or_else(|| panic!("field {field_name} type must be object"));
2822
2823 assert_eq!(ty.get("type").and_then(Value::as_str), Some("fixed"));
2824 assert_eq!(
2825 ty.get("logicalType").and_then(Value::as_str),
2826 Some(logical_type)
2827 );
2828 assert_eq!(
2829 ty.get("namespace").and_then(Value::as_str),
2830 Some(namespace),
2831 "field {field_name} must preserve avro.namespace metadata"
2832 );
2833 }
2834 }
2835
2836 #[cfg(feature = "avro_custom_types")]
2837 #[test]
2838 fn test_custom_fixed_logical_types_omit_namespace_without_metadata() {
2839 let mut md_u64 = HashMap::new();
2840 md_u64.insert(AVRO_NAME_METADATA_KEY.to_string(), "U64Type".to_string());
2841
2842 let mut md_f16 = HashMap::new();
2843 md_f16.insert(AVRO_NAME_METADATA_KEY.to_string(), "F16Type".to_string());
2844
2845 let mut md_iv_ym = HashMap::new();
2846 md_iv_ym.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvYmType".to_string());
2847
2848 let mut md_iv_dt = HashMap::new();
2849 md_iv_dt.insert(AVRO_NAME_METADATA_KEY.to_string(), "IvDtType".to_string());
2850
2851 let arrow_schema = ArrowSchema::new(vec![
2852 ArrowField::new("u64_col", DataType::UInt64, false).with_metadata(md_u64),
2853 ArrowField::new("f16_col", DataType::Float16, false).with_metadata(md_f16),
2854 ArrowField::new(
2855 "iv_ym_col",
2856 DataType::Interval(IntervalUnit::YearMonth),
2857 false,
2858 )
2859 .with_metadata(md_iv_ym),
2860 ArrowField::new(
2861 "iv_dt_col",
2862 DataType::Interval(IntervalUnit::DayTime),
2863 false,
2864 )
2865 .with_metadata(md_iv_dt),
2866 ]);
2867
2868 let avro = AvroSchema::try_from(&arrow_schema).unwrap();
2869 let root: Value = serde_json::from_str(&avro.json_string).unwrap();
2870 let fields = root
2871 .get("fields")
2872 .and_then(|f| f.as_array())
2873 .expect("record fields array");
2874
2875 for field_name in ["u64_col", "f16_col", "iv_ym_col", "iv_dt_col"] {
2876 let field = fields
2877 .iter()
2878 .find(|f| f.get("name").and_then(Value::as_str) == Some(field_name))
2879 .unwrap_or_else(|| panic!("missing field {field_name}"));
2880 let ty = field
2881 .get("type")
2882 .and_then(Value::as_object)
2883 .unwrap_or_else(|| panic!("field {field_name} type must be object"));
2884
2885 assert_eq!(ty.get("type").and_then(Value::as_str), Some("fixed"));
2886 assert!(
2887 !ty.contains_key("namespace"),
2888 "field {field_name} should not include namespace when metadata lacks avro.namespace"
2889 );
2890 }
2891 }
2892
2893 #[test]
2894 fn test_temporal_mappings() {
2895 let cases = vec![
2896 (DataType::Date32, "\"logicalType\":\"date\""),
2897 (
2898 DataType::Time32(TimeUnit::Millisecond),
2899 "\"logicalType\":\"time-millis\"",
2900 ),
2901 (
2902 DataType::Time64(TimeUnit::Microsecond),
2903 "\"logicalType\":\"time-micros\"",
2904 ),
2905 (
2906 DataType::Timestamp(TimeUnit::Millisecond, None),
2907 "\"logicalType\":\"local-timestamp-millis\"",
2908 ),
2909 (
2910 DataType::Timestamp(TimeUnit::Microsecond, Some("+00:00".into())),
2911 "\"logicalType\":\"timestamp-micros\"",
2912 ),
2913 ];
2914 for (dt, needle) in cases {
2915 let field = ArrowField::new("ts", dt.clone(), true);
2916 let arrow_schema = single_field_schema(field);
2917 let avro = AvroSchema::try_from(&arrow_schema).unwrap();
2918 assert_json_contains(&avro.json_string, needle);
2919 }
2920 }
2921
2922 #[test]
2923 fn test_decimal_and_uuid() {
2924 let decimal_field = ArrowField::new("amount", DataType::Decimal128(25, 2), false);
2925 let dec_schema = single_field_schema(decimal_field);
2926 let avro_dec = AvroSchema::try_from(&dec_schema).unwrap();
2927 assert_json_contains(&avro_dec.json_string, "\"logicalType\":\"decimal\"");
2928 assert_json_contains(&avro_dec.json_string, "\"precision\":25");
2929 assert_json_contains(&avro_dec.json_string, "\"scale\":2");
2930 let mut md = HashMap::new();
2931 md.insert("logicalType".into(), "uuid".into());
2932 let uuid_field =
2933 ArrowField::new("id", DataType::FixedSizeBinary(16), false).with_metadata(md);
2934 let uuid_schema = single_field_schema(uuid_field);
2935 let avro_uuid = AvroSchema::try_from(&uuid_schema).unwrap();
2936 assert_json_contains(&avro_uuid.json_string, "\"logicalType\":\"uuid\"");
2937 }
2938
2939 #[cfg(not(feature = "avro_custom_types"))]
2940 #[test]
2941 fn test_interval_month_day_nano_duration_schema() {
2942 let interval_field = ArrowField::new(
2943 "span",
2944 DataType::Interval(IntervalUnit::MonthDayNano),
2945 false,
2946 );
2947 let s = single_field_schema(interval_field);
2948 let avro = AvroSchema::try_from(&s).unwrap();
2949 assert_json_contains(&avro.json_string, "\"logicalType\":\"duration\"");
2950 assert_json_contains(&avro.json_string, "\"size\":12");
2951 }
2952
2953 #[cfg(feature = "avro_custom_types")]
2954 #[test]
2955 fn test_interval_month_day_nano_custom_schema() {
2956 let interval_field = ArrowField::new(
2957 "span",
2958 DataType::Interval(IntervalUnit::MonthDayNano),
2959 false,
2960 );
2961 let s = single_field_schema(interval_field);
2962 let avro = AvroSchema::try_from(&s).unwrap();
2963 assert_json_contains(
2964 &avro.json_string,
2965 "\"logicalType\":\"arrow.interval-month-day-nano\"",
2966 );
2967 assert_json_contains(&avro.json_string, "\"size\":16");
2968 }
2969
2970 #[cfg(feature = "avro_custom_types")]
2971 #[test]
2972 fn test_duration_custom_logical_type() {
2973 let dur_field = ArrowField::new("latency", DataType::Duration(TimeUnit::Nanosecond), false);
2974 let s2 = single_field_schema(dur_field);
2975 let avro2 = AvroSchema::try_from(&s2).unwrap();
2976 assert_json_contains(
2977 &avro2.json_string,
2978 "\"logicalType\":\"arrow.duration-nanos\"",
2979 );
2980 }
2981
2982 #[test]
2983 fn test_complex_types() {
2984 let list_dt = DataType::List(Arc::new(ArrowField::new("item", DataType::Int32, true)));
2985 let list_schema = single_field_schema(ArrowField::new("numbers", list_dt, false));
2986 let avro_list = AvroSchema::try_from(&list_schema).unwrap();
2987 assert_json_contains(&avro_list.json_string, "\"type\":\"array\"");
2988 assert_json_contains(&avro_list.json_string, "\"items\"");
2989 let value_field = ArrowField::new("value", DataType::Boolean, true);
2990 let entries_struct = ArrowField::new(
2991 "entries",
2992 DataType::Struct(Fields::from(vec![
2993 ArrowField::new("key", DataType::Utf8, false),
2994 value_field.clone(),
2995 ])),
2996 false,
2997 );
2998 let map_dt = DataType::Map(Arc::new(entries_struct), false);
2999 let map_schema = single_field_schema(ArrowField::new("props", map_dt, false));
3000 let avro_map = AvroSchema::try_from(&map_schema).unwrap();
3001 assert_json_contains(&avro_map.json_string, "\"type\":\"map\"");
3002 assert_json_contains(&avro_map.json_string, "\"values\"");
3003 let struct_dt = DataType::Struct(Fields::from(vec![
3004 ArrowField::new("f1", DataType::Int64, false),
3005 ArrowField::new("f2", DataType::Utf8, true),
3006 ]));
3007 let struct_schema = single_field_schema(ArrowField::new("person", struct_dt, true));
3008 let avro_struct = AvroSchema::try_from(&struct_schema).unwrap();
3009 assert_json_contains(&avro_struct.json_string, "\"type\":\"record\"");
3010 assert_json_contains(&avro_struct.json_string, "\"null\"");
3011 }
3012
3013 #[test]
3014 fn test_enum_dictionary() {
3015 let mut md = HashMap::new();
3016 md.insert(
3017 AVRO_ENUM_SYMBOLS_METADATA_KEY.into(),
3018 "[\"OPEN\",\"CLOSED\"]".into(),
3019 );
3020 let enum_dt = DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
3021 let field = ArrowField::new("status", enum_dt, false).with_metadata(md);
3022 let schema = single_field_schema(field);
3023 let avro = AvroSchema::try_from(&schema).unwrap();
3024 assert_json_contains(&avro.json_string, "\"type\":\"enum\"");
3025 assert_json_contains(&avro.json_string, "\"symbols\":[\"OPEN\",\"CLOSED\"]");
3026 }
3027
3028 #[test]
3029 fn test_run_end_encoded() {
3030 let ree_dt = DataType::RunEndEncoded(
3031 Arc::new(ArrowField::new("run_ends", DataType::Int32, false)),
3032 Arc::new(ArrowField::new("values", DataType::Utf8, false)),
3033 );
3034 let s = single_field_schema(ArrowField::new("text", ree_dt, false));
3035 let avro = AvroSchema::try_from(&s).unwrap();
3036 assert_json_contains(&avro.json_string, "\"string\"");
3037 }
3038
3039 #[test]
3040 fn test_dense_union() {
3041 let uf: UnionFields = vec![
3042 (2i8, Arc::new(ArrowField::new("a", DataType::Int32, false))),
3043 (7i8, Arc::new(ArrowField::new("b", DataType::Utf8, true))),
3044 ]
3045 .into_iter()
3046 .collect();
3047 let union_dt = DataType::Union(uf, UnionMode::Dense);
3048 let s = single_field_schema(ArrowField::new("u", union_dt, false));
3049 let avro =
3050 AvroSchema::try_from(&s).expect("Arrow Union -> Avro union conversion should succeed");
3051 let v: serde_json::Value = serde_json::from_str(&avro.json_string).unwrap();
3052 let fields = v
3053 .get("fields")
3054 .and_then(|x| x.as_array())
3055 .expect("fields array");
3056 let u_field = fields
3057 .iter()
3058 .find(|f| f.get("name").and_then(|n| n.as_str()) == Some("u"))
3059 .expect("field 'u'");
3060 let union = u_field.get("type").expect("u.type");
3061 let arr = union.as_array().expect("u.type must be Avro union array");
3062 assert_eq!(arr.len(), 2, "expected two union branches");
3063 let first = &arr[0];
3064 let obj = first
3065 .as_object()
3066 .expect("first branch should be an object with metadata");
3067 assert_eq!(obj.get("type").and_then(|t| t.as_str()), Some("int"));
3068 assert_eq!(
3069 obj.get("arrowUnionMode").and_then(|m| m.as_str()),
3070 Some("dense")
3071 );
3072 let type_ids: Vec<i64> = obj
3073 .get("arrowUnionTypeIds")
3074 .and_then(|a| a.as_array())
3075 .expect("arrowUnionTypeIds array")
3076 .iter()
3077 .map(|n| n.as_i64().expect("i64"))
3078 .collect();
3079 assert_eq!(type_ids, vec![2, 7], "type id ordering should be preserved");
3080 assert_eq!(arr[1], Value::String("string".into()));
3081 }
3082
3083 #[test]
3084 fn round_trip_primitive() {
3085 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("f1", DataType::Int32, false)]);
3086 let avro_schema = AvroSchema::try_from(&arrow_schema).unwrap();
3087 let decoded = avro_schema.schema().unwrap();
3088 assert!(matches!(decoded, Schema::Complex(_)));
3089 }
3090
3091 #[test]
3092 fn test_name_generator_sanitization_and_uniqueness() {
3093 let f1 = ArrowField::new("weird-name", DataType::FixedSizeBinary(8), false);
3094 let f2 = ArrowField::new("weird name", DataType::FixedSizeBinary(8), false);
3095 let f3 = ArrowField::new("123bad", DataType::FixedSizeBinary(8), false);
3096 let arrow_schema = ArrowSchema::new(vec![f1, f2, f3]);
3097 let avro = AvroSchema::try_from(&arrow_schema).unwrap();
3098 assert_json_contains(&avro.json_string, "\"name\":\"weird_name\"");
3099 assert_json_contains(&avro.json_string, "\"name\":\"weird_name_1\"");
3100 assert_json_contains(&avro.json_string, "\"name\":\"_123bad\"");
3101 }
3102
3103 #[cfg(not(feature = "avro_custom_types"))]
3104 #[test]
3105 fn test_date64_logical_type_mapping() {
3106 let field = ArrowField::new("d", DataType::Date64, true);
3107 let schema = single_field_schema(field);
3108 let avro = AvroSchema::try_from(&schema).unwrap();
3109 assert_json_contains(
3110 &avro.json_string,
3111 "\"logicalType\":\"local-timestamp-millis\"",
3112 );
3113 }
3114
3115 #[cfg(feature = "avro_custom_types")]
3116 #[test]
3117 fn test_date64_logical_type_mapping_custom() {
3118 let field = ArrowField::new("d", DataType::Date64, true);
3119 let schema = single_field_schema(field);
3120 let avro = AvroSchema::try_from(&schema).unwrap();
3121 assert_json_contains(&avro.json_string, "\"logicalType\":\"arrow.date64\"");
3122 }
3123
3124 #[cfg(feature = "avro_custom_types")]
3125 #[test]
3126 fn test_duration_list_extras_propagated() {
3127 let child = ArrowField::new("lat", DataType::Duration(TimeUnit::Microsecond), false);
3128 let list_dt = DataType::List(Arc::new(child));
3129 let arrow_schema = single_field_schema(ArrowField::new("durations", list_dt, false));
3130 let avro = AvroSchema::try_from(&arrow_schema).unwrap();
3131 assert_json_contains(
3132 &avro.json_string,
3133 "\"logicalType\":\"arrow.duration-micros\"",
3134 );
3135 }
3136
3137 #[cfg(not(feature = "avro_custom_types"))]
3138 #[test]
3139 fn test_interval_yearmonth_extra() {
3140 let field = ArrowField::new("iv", DataType::Interval(IntervalUnit::YearMonth), false);
3141 let schema = single_field_schema(field);
3142 let avro = AvroSchema::try_from(&schema).unwrap();
3143 assert_json_contains(&avro.json_string, "\"arrowIntervalUnit\":\"yearmonth\"");
3144 }
3145
3146 #[cfg(not(feature = "avro_custom_types"))]
3147 #[test]
3148 fn test_interval_daytime_extra() {
3149 let field = ArrowField::new("iv_dt", DataType::Interval(IntervalUnit::DayTime), false);
3150 let schema = single_field_schema(field);
3151 let avro = AvroSchema::try_from(&schema).unwrap();
3152 assert_json_contains(&avro.json_string, "\"arrowIntervalUnit\":\"daytime\"");
3153 }
3154
3155 #[cfg(feature = "avro_custom_types")]
3156 #[test]
3157 fn test_interval_yearmonth_custom() {
3158 let field = ArrowField::new("iv", DataType::Interval(IntervalUnit::YearMonth), false);
3159 let schema = single_field_schema(field);
3160 let avro = AvroSchema::try_from(&schema).unwrap();
3161 assert_json_contains(
3162 &avro.json_string,
3163 "\"logicalType\":\"arrow.interval-year-month\"",
3164 );
3165 }
3166
3167 #[cfg(feature = "avro_custom_types")]
3168 #[test]
3169 fn test_interval_daytime_custom() {
3170 let field = ArrowField::new("iv_dt", DataType::Interval(IntervalUnit::DayTime), false);
3171 let schema = single_field_schema(field);
3172 let avro = AvroSchema::try_from(&schema).unwrap();
3173 assert_json_contains(
3174 &avro.json_string,
3175 "\"logicalType\":\"arrow.interval-day-time\"",
3176 );
3177 }
3178
3179 #[test]
3180 fn test_fixed_size_list_extra() {
3181 let child = ArrowField::new("item", DataType::Int32, false);
3182 let dt = DataType::FixedSizeList(Arc::new(child), 3);
3183 let schema = single_field_schema(ArrowField::new("triples", dt, false));
3184 let avro = AvroSchema::try_from(&schema).unwrap();
3185 assert_json_contains(&avro.json_string, "\"arrowFixedSize\":3");
3186 }
3187
3188 #[cfg(feature = "avro_custom_types")]
3189 #[test]
3190 fn test_map_duration_value_extra() {
3191 let val_field = ArrowField::new("value", DataType::Duration(TimeUnit::Second), true);
3192 let entries_struct = ArrowField::new(
3193 "entries",
3194 DataType::Struct(Fields::from(vec![
3195 ArrowField::new("key", DataType::Utf8, false),
3196 val_field,
3197 ])),
3198 false,
3199 );
3200 let map_dt = DataType::Map(Arc::new(entries_struct), false);
3201 let schema = single_field_schema(ArrowField::new("metrics", map_dt, false));
3202 let avro = AvroSchema::try_from(&schema).unwrap();
3203 assert_json_contains(
3204 &avro.json_string,
3205 "\"logicalType\":\"arrow.duration-seconds\"",
3206 );
3207 }
3208
3209 #[test]
3210 fn test_schema_with_non_string_defaults_decodes_successfully() {
3211 let schema_json = r#"{
3212 "type": "record",
3213 "name": "R",
3214 "fields": [
3215 {"name": "a", "type": "int", "default": 0},
3216 {"name": "b", "type": {"type": "array", "items": "long"}, "default": [1, 2, 3]},
3217 {"name": "c", "type": {"type": "map", "values": "double"}, "default": {"x": 1.5, "y": 2.5}},
3218 {"name": "inner", "type": {"type": "record", "name": "Inner", "fields": [
3219 {"name": "flag", "type": "boolean", "default": true},
3220 {"name": "name", "type": "string", "default": "hi"}
3221 ]}, "default": {"flag": false, "name": "d"}},
3222 {"name": "u", "type": ["int", "null"], "default": 42}
3223 ]
3224 }"#;
3225 let schema: Schema = serde_json::from_str(schema_json).expect("schema should parse");
3226 match &schema {
3227 Schema::Complex(ComplexType::Record(_)) => {}
3228 other => panic!("expected record schema, got: {:?}", other),
3229 }
3230 let field = crate::codec::AvroField::try_from(&schema)
3232 .expect("Avro->Arrow conversion should succeed");
3233 let arrow_field = field.field();
3234 let expected_list_item = ArrowField::new(
3236 arrow_schema::Field::LIST_FIELD_DEFAULT_NAME,
3237 DataType::Int64,
3238 false,
3239 );
3240 let expected_b = ArrowField::new("b", DataType::List(Arc::new(expected_list_item)), false);
3241
3242 let expected_map_value = ArrowField::new("value", DataType::Float64, false);
3243 let expected_entries = ArrowField::new(
3244 "entries",
3245 DataType::Struct(Fields::from(vec![
3246 ArrowField::new("key", DataType::Utf8, false),
3247 expected_map_value,
3248 ])),
3249 false,
3250 );
3251 let expected_c =
3252 ArrowField::new("c", DataType::Map(Arc::new(expected_entries), false), false);
3253 let mut inner_md = std::collections::HashMap::new();
3254 inner_md.insert(AVRO_NAME_METADATA_KEY.to_string(), "Inner".to_string());
3255 let expected_inner = ArrowField::new(
3256 "inner",
3257 DataType::Struct(Fields::from(vec![
3258 ArrowField::new("flag", DataType::Boolean, false),
3259 ArrowField::new("name", DataType::Utf8, false),
3260 ])),
3261 false,
3262 )
3263 .with_metadata(inner_md);
3264 let mut root_md = std::collections::HashMap::new();
3265 root_md.insert(AVRO_NAME_METADATA_KEY.to_string(), "R".to_string());
3266 let expected = ArrowField::new(
3267 "R",
3268 DataType::Struct(Fields::from(vec![
3269 ArrowField::new("a", DataType::Int32, false),
3270 expected_b,
3271 expected_c,
3272 expected_inner,
3273 ArrowField::new("u", DataType::Int32, true),
3274 ])),
3275 false,
3276 )
3277 .with_metadata(root_md);
3278 assert_eq!(arrow_field, expected);
3279 }
3280
3281 #[test]
3282 fn default_order_is_consistent() {
3283 let arrow_schema = ArrowSchema::new(vec![ArrowField::new("s", DataType::Utf8, true)]);
3284 let a = AvroSchema::try_from(&arrow_schema).unwrap().json_string;
3285 let b = AvroSchema::from_arrow_with_options(&arrow_schema, None);
3286 assert_eq!(a, b.unwrap().json_string);
3287 }
3288
3289 #[test]
3290 fn test_union_branch_missing_name_errors() {
3291 for t in ["record", "enum", "fixed"] {
3292 let branch = json!({ "type": t });
3293 let err = union_branch_signature(&branch).unwrap_err().to_string();
3294 assert!(
3295 err.contains(&format!("Union branch '{t}' missing required 'name'")),
3296 "expected missing-name error for {t}, got: {err}"
3297 );
3298 }
3299 }
3300
3301 #[test]
3302 fn test_union_branch_named_type_signature_includes_name() {
3303 let rec = json!({ "type": "record", "name": "Foo" });
3304 assert_eq!(union_branch_signature(&rec).unwrap(), "N:record:Foo");
3305 let en = json!({ "type": "enum", "name": "Color", "symbols": ["R", "G", "B"] });
3306 assert_eq!(union_branch_signature(&en).unwrap(), "N:enum:Color");
3307 let fx = json!({ "type": "fixed", "name": "Bytes16", "size": 16 });
3308 assert_eq!(union_branch_signature(&fx).unwrap(), "N:fixed:Bytes16");
3309 }
3310
3311 #[test]
3312 fn test_record_field_alias_resolution_without_default() {
3313 let writer_json = r#"{
3314 "type":"record",
3315 "name":"R",
3316 "fields":[{"name":"old","type":"int"}]
3317 }"#;
3318 let reader_json = r#"{
3319 "type":"record",
3320 "name":"R",
3321 "fields":[{"name":"new","aliases":["old"],"type":"int"}]
3322 }"#;
3323 let writer: Schema = serde_json::from_str(writer_json).unwrap();
3324 let reader: Schema = serde_json::from_str(reader_json).unwrap();
3325 let resolved = AvroFieldBuilder::new(&writer)
3326 .with_reader_schema(&reader)
3327 .with_utf8view(false)
3328 .with_strict_mode(false)
3329 .build()
3330 .unwrap();
3331 let expected = ArrowField::new(
3332 "R",
3333 DataType::Struct(Fields::from(vec![ArrowField::new(
3334 "new",
3335 DataType::Int32,
3336 false,
3337 )])),
3338 false,
3339 )
3340 .with_metadata(HashMap::from_iter([(
3341 "avro.name".to_owned(),
3342 "R".to_owned(),
3343 )]));
3344 assert_eq!(resolved.field(), expected);
3345 }
3346
3347 #[test]
3348 fn test_record_field_alias_ambiguous_in_strict_mode_errors() {
3349 let writer_json = r#"{
3350 "type":"record",
3351 "name":"R",
3352 "fields":[
3353 {"name":"a","type":"int","aliases":["old"]},
3354 {"name":"b","type":"int","aliases":["old"]}
3355 ]
3356 }"#;
3357 let reader_json = r#"{
3358 "type":"record",
3359 "name":"R",
3360 "fields":[{"name":"target","type":"int","aliases":["old"]}]
3361 }"#;
3362 let writer: Schema = serde_json::from_str(writer_json).unwrap();
3363 let reader: Schema = serde_json::from_str(reader_json).unwrap();
3364 let err = AvroFieldBuilder::new(&writer)
3365 .with_reader_schema(&reader)
3366 .with_utf8view(false)
3367 .with_strict_mode(true)
3368 .build()
3369 .unwrap_err()
3370 .to_string();
3371 assert!(
3372 err.contains("Ambiguous alias 'old'"),
3373 "expected ambiguous-alias error, got: {err}"
3374 );
3375 }
3376
3377 #[test]
3378 fn test_pragmatic_writer_field_alias_mapping_non_strict() {
3379 let writer_json = r#"{
3380 "type":"record",
3381 "name":"R",
3382 "fields":[{"name":"before","type":"int","aliases":["now"]}]
3383 }"#;
3384 let reader_json = r#"{
3385 "type":"record",
3386 "name":"R",
3387 "fields":[{"name":"now","type":"int"}]
3388 }"#;
3389 let writer: Schema = serde_json::from_str(writer_json).unwrap();
3390 let reader: Schema = serde_json::from_str(reader_json).unwrap();
3391 let resolved = AvroFieldBuilder::new(&writer)
3392 .with_reader_schema(&reader)
3393 .with_utf8view(false)
3394 .with_strict_mode(false)
3395 .build()
3396 .unwrap();
3397 let expected = ArrowField::new(
3398 "R",
3399 DataType::Struct(Fields::from(vec![ArrowField::new(
3400 "now",
3401 DataType::Int32,
3402 false,
3403 )])),
3404 false,
3405 )
3406 .with_metadata(HashMap::from_iter([(
3407 "avro.name".to_owned(),
3408 "R".to_owned(),
3409 )]));
3410 assert_eq!(resolved.field(), expected);
3411 }
3412
3413 #[test]
3414 fn test_missing_reader_field_null_first_no_default_is_ok() {
3415 let writer_json = r#"{
3416 "type":"record",
3417 "name":"R",
3418 "fields":[{"name":"a","type":"int"}]
3419 }"#;
3420 let reader_json = r#"{
3421 "type":"record",
3422 "name":"R",
3423 "fields":[
3424 {"name":"a","type":"int"},
3425 {"name":"b","type":["null","int"]}
3426 ]
3427 }"#;
3428 let writer: Schema = serde_json::from_str(writer_json).unwrap();
3429 let reader: Schema = serde_json::from_str(reader_json).unwrap();
3430 let resolved = AvroFieldBuilder::new(&writer)
3431 .with_reader_schema(&reader)
3432 .with_utf8view(false)
3433 .with_strict_mode(false)
3434 .build()
3435 .unwrap();
3436 let expected = ArrowField::new(
3437 "R",
3438 DataType::Struct(Fields::from(vec![
3439 ArrowField::new("a", DataType::Int32, false),
3440 ArrowField::new("b", DataType::Int32, true).with_metadata(HashMap::from([(
3441 AVRO_FIELD_DEFAULT_METADATA_KEY.to_string(),
3442 "null".to_string(),
3443 )])),
3444 ])),
3445 false,
3446 )
3447 .with_metadata(HashMap::from_iter([(
3448 "avro.name".to_owned(),
3449 "R".to_owned(),
3450 )]));
3451 assert_eq!(resolved.field(), expected);
3452 }
3453
3454 #[test]
3455 fn test_missing_reader_field_null_second_without_default_errors() {
3456 let writer_json = r#"{
3457 "type":"record",
3458 "name":"R",
3459 "fields":[{"name":"a","type":"int"}]
3460 }"#;
3461 let reader_json = r#"{
3462 "type":"record",
3463 "name":"R",
3464 "fields":[
3465 {"name":"a","type":"int"},
3466 {"name":"b","type":["int","null"]}
3467 ]
3468 }"#;
3469 let writer: Schema = serde_json::from_str(writer_json).unwrap();
3470 let reader: Schema = serde_json::from_str(reader_json).unwrap();
3471 let err = AvroFieldBuilder::new(&writer)
3472 .with_reader_schema(&reader)
3473 .with_utf8view(false)
3474 .with_strict_mode(false)
3475 .build()
3476 .unwrap_err()
3477 .to_string();
3478 assert!(
3479 err.contains("must have a default value"),
3480 "expected missing-default error, got: {err}"
3481 );
3482 }
3483
3484 #[test]
3485 fn test_from_arrow_with_options_respects_schema_metadata_when_not_stripping() {
3486 let field = ArrowField::new("x", DataType::Int32, true);
3487 let injected_json =
3488 r#"{"type":"record","name":"Injected","fields":[{"name":"ignored","type":"int"}]}"#
3489 .to_string();
3490 let mut md = HashMap::new();
3491 md.insert(SCHEMA_METADATA_KEY.to_string(), injected_json.clone());
3492 md.insert("custom".to_string(), "123".to_string());
3493 let arrow_schema = ArrowSchema::new_with_metadata(vec![field], md);
3494 let opts = AvroSchemaOptions {
3495 null_order: Some(Nullability::NullSecond),
3496 strip_metadata: false,
3497 };
3498 let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
3499 assert_eq!(
3500 out.json_string, injected_json,
3501 "When strip_metadata=false and avro.schema is present, return the embedded JSON verbatim"
3502 );
3503 let v: Value = serde_json::from_str(&out.json_string).unwrap();
3504 assert_eq!(v.get("type").and_then(|t| t.as_str()), Some("record"));
3505 assert_eq!(v.get("name").and_then(|n| n.as_str()), Some("Injected"));
3506 }
3507
3508 #[test]
3509 fn test_from_arrow_with_options_ignores_schema_metadata_when_stripping_and_keeps_passthrough() {
3510 let field = ArrowField::new("x", DataType::Int32, true);
3511 let injected_json =
3512 r#"{"type":"record","name":"Injected","fields":[{"name":"ignored","type":"int"}]}"#
3513 .to_string();
3514 let mut md = HashMap::new();
3515 md.insert(SCHEMA_METADATA_KEY.to_string(), injected_json);
3516 md.insert("custom_meta".to_string(), "7".to_string());
3517 let arrow_schema = ArrowSchema::new_with_metadata(vec![field], md);
3518 let opts = AvroSchemaOptions {
3519 null_order: Some(Nullability::NullFirst),
3520 strip_metadata: true,
3521 };
3522 let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
3523 assert_json_contains(&out.json_string, "\"type\":\"record\"");
3524 assert_json_contains(&out.json_string, "\"name\":\"topLevelRecord\"");
3525 assert_json_contains(&out.json_string, "\"custom_meta\":7");
3526 }
3527
3528 #[test]
3529 fn test_from_arrow_with_options_null_first_for_nullable_primitive() {
3530 let field = ArrowField::new("s", DataType::Utf8, true);
3531 let arrow_schema = single_field_schema(field);
3532 let opts = AvroSchemaOptions {
3533 null_order: Some(Nullability::NullFirst),
3534 strip_metadata: true,
3535 };
3536 let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
3537 let v: Value = serde_json::from_str(&out.json_string).unwrap();
3538 let arr = v["fields"][0]["type"]
3539 .as_array()
3540 .expect("nullable primitive should be Avro union array");
3541 assert_eq!(arr[0], Value::String("null".into()));
3542 assert_eq!(arr[1], Value::String("string".into()));
3543 }
3544
3545 #[test]
3546 fn test_from_arrow_with_options_null_second_for_nullable_primitive() {
3547 let field = ArrowField::new("s", DataType::Utf8, true);
3548 let arrow_schema = single_field_schema(field);
3549 let opts = AvroSchemaOptions {
3550 null_order: Some(Nullability::NullSecond),
3551 strip_metadata: true,
3552 };
3553 let out = AvroSchema::from_arrow_with_options(&arrow_schema, Some(opts)).unwrap();
3554 let v: Value = serde_json::from_str(&out.json_string).unwrap();
3555 let arr = v["fields"][0]["type"]
3556 .as_array()
3557 .expect("nullable primitive should be Avro union array");
3558 assert_eq!(arr[0], Value::String("string".into()));
3559 assert_eq!(arr[1], Value::String("null".into()));
3560 }
3561
3562 #[test]
3563 fn test_from_arrow_with_options_union_extras_respected_by_strip_metadata() {
3564 let uf: UnionFields = vec![
3565 (2i8, Arc::new(ArrowField::new("a", DataType::Int32, false))),
3566 (7i8, Arc::new(ArrowField::new("b", DataType::Utf8, false))),
3567 ]
3568 .into_iter()
3569 .collect();
3570 let union_dt = DataType::Union(uf, UnionMode::Dense);
3571 let arrow_schema = single_field_schema(ArrowField::new("u", union_dt, true));
3572 let with_extras = AvroSchema::from_arrow_with_options(
3573 &arrow_schema,
3574 Some(AvroSchemaOptions {
3575 null_order: Some(Nullability::NullFirst),
3576 strip_metadata: false,
3577 }),
3578 )
3579 .unwrap();
3580 let v_with: Value = serde_json::from_str(&with_extras.json_string).unwrap();
3581 let union_arr = v_with["fields"][0]["type"].as_array().expect("union array");
3582 let first_obj = union_arr
3583 .iter()
3584 .find(|b| b.is_object())
3585 .expect("expected an object branch with extras");
3586 let obj = first_obj.as_object().unwrap();
3587 assert_eq!(obj.get("type").and_then(|t| t.as_str()), Some("int"));
3588 assert_eq!(
3589 obj.get("arrowUnionMode").and_then(|m| m.as_str()),
3590 Some("dense")
3591 );
3592 let type_ids: Vec<i64> = obj["arrowUnionTypeIds"]
3593 .as_array()
3594 .expect("arrowUnionTypeIds array")
3595 .iter()
3596 .map(|n| n.as_i64().expect("i64"))
3597 .collect();
3598 assert_eq!(type_ids, vec![2, 7]);
3599 let stripped = AvroSchema::from_arrow_with_options(
3600 &arrow_schema,
3601 Some(AvroSchemaOptions {
3602 null_order: Some(Nullability::NullFirst),
3603 strip_metadata: true,
3604 }),
3605 )
3606 .unwrap();
3607 let v_stripped: Value = serde_json::from_str(&stripped.json_string).unwrap();
3608 let union_arr2 = v_stripped["fields"][0]["type"]
3609 .as_array()
3610 .expect("union array");
3611 assert!(
3612 !union_arr2.iter().any(|b| b
3613 .as_object()
3614 .is_some_and(|m| m.contains_key("arrowUnionMode"))),
3615 "extras must be removed when strip_metadata=true"
3616 );
3617 assert_eq!(union_arr2[0], Value::String("null".into()));
3618 assert_eq!(union_arr2[1], Value::String("int".into()));
3619 assert_eq!(union_arr2[2], Value::String("string".into()));
3620 }
3621
3622 #[test]
3623 fn test_project_empty_projection() {
3624 let schema_json = r#"{
3625 "type": "record",
3626 "name": "Test",
3627 "fields": [
3628 {"name": "a", "type": "int"},
3629 {"name": "b", "type": "string"}
3630 ]
3631 }"#;
3632 let schema = AvroSchema::new(schema_json.to_string());
3633 let projected = schema.project(&[]).unwrap();
3634 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3635 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3636 assert!(
3637 fields.is_empty(),
3638 "Empty projection should yield empty fields"
3639 );
3640 }
3641
3642 #[test]
3643 fn test_project_single_field() {
3644 let schema_json = r#"{
3645 "type": "record",
3646 "name": "Test",
3647 "fields": [
3648 {"name": "a", "type": "int"},
3649 {"name": "b", "type": "string"},
3650 {"name": "c", "type": "long"}
3651 ]
3652 }"#;
3653 let schema = AvroSchema::new(schema_json.to_string());
3654 let projected = schema.project(&[1]).unwrap();
3655 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3656 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3657 assert_eq!(fields.len(), 1);
3658 assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("b"));
3659 }
3660
3661 #[test]
3662 fn test_project_multiple_fields() {
3663 let schema_json = r#"{
3664 "type": "record",
3665 "name": "Test",
3666 "fields": [
3667 {"name": "a", "type": "int"},
3668 {"name": "b", "type": "string"},
3669 {"name": "c", "type": "long"},
3670 {"name": "d", "type": "boolean"}
3671 ]
3672 }"#;
3673 let schema = AvroSchema::new(schema_json.to_string());
3674 let projected = schema.project(&[0, 2, 3]).unwrap();
3675 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3676 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3677 assert_eq!(fields.len(), 3);
3678 assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("a"));
3679 assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("c"));
3680 assert_eq!(fields[2].get("name").and_then(|n| n.as_str()), Some("d"));
3681 }
3682
3683 #[test]
3684 fn test_project_all_fields() {
3685 let schema_json = r#"{
3686 "type": "record",
3687 "name": "Test",
3688 "fields": [
3689 {"name": "a", "type": "int"},
3690 {"name": "b", "type": "string"}
3691 ]
3692 }"#;
3693 let schema = AvroSchema::new(schema_json.to_string());
3694 let projected = schema.project(&[0, 1]).unwrap();
3695 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3696 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3697 assert_eq!(fields.len(), 2);
3698 assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("a"));
3699 assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("b"));
3700 }
3701
3702 #[test]
3703 fn test_project_reorder_fields() {
3704 let schema_json = r#"{
3705 "type": "record",
3706 "name": "Test",
3707 "fields": [
3708 {"name": "a", "type": "int"},
3709 {"name": "b", "type": "string"},
3710 {"name": "c", "type": "long"}
3711 ]
3712 }"#;
3713 let schema = AvroSchema::new(schema_json.to_string());
3714 let projected = schema.project(&[2, 0, 1]).unwrap();
3716 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3717 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3718 assert_eq!(fields.len(), 3);
3719 assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("c"));
3720 assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("a"));
3721 assert_eq!(fields[2].get("name").and_then(|n| n.as_str()), Some("b"));
3722 }
3723
3724 #[test]
3725 fn test_project_preserves_record_metadata() {
3726 let schema_json = r#"{
3727 "type": "record",
3728 "name": "MyRecord",
3729 "namespace": "com.example",
3730 "doc": "A test record",
3731 "aliases": ["OldRecord"],
3732 "fields": [
3733 {"name": "a", "type": "int"},
3734 {"name": "b", "type": "string"}
3735 ]
3736 }"#;
3737 let schema = AvroSchema::new(schema_json.to_string());
3738 let projected = schema.project(&[0]).unwrap();
3739 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3740 assert_eq!(v.get("name").and_then(|n| n.as_str()), Some("MyRecord"));
3741 assert_eq!(
3742 v.get("namespace").and_then(|n| n.as_str()),
3743 Some("com.example")
3744 );
3745 assert_eq!(v.get("doc").and_then(|n| n.as_str()), Some("A test record"));
3746 assert!(v.get("aliases").is_some());
3747 }
3748
3749 #[test]
3750 fn test_project_preserves_field_metadata() {
3751 let schema_json = r#"{
3752 "type": "record",
3753 "name": "Test",
3754 "fields": [
3755 {"name": "a", "type": "int", "doc": "Field A", "default": 0},
3756 {"name": "b", "type": "string"}
3757 ]
3758 }"#;
3759 let schema = AvroSchema::new(schema_json.to_string());
3760 let projected = schema.project(&[0]).unwrap();
3761 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3762 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3763 assert_eq!(
3764 fields[0].get("doc").and_then(|d| d.as_str()),
3765 Some("Field A")
3766 );
3767 assert_eq!(fields[0].get("default").and_then(|d| d.as_i64()), Some(0));
3768 }
3769
3770 #[test]
3771 fn test_project_with_nested_record() {
3772 let schema_json = r#"{
3773 "type": "record",
3774 "name": "Outer",
3775 "fields": [
3776 {"name": "id", "type": "int"},
3777 {"name": "inner", "type": {
3778 "type": "record",
3779 "name": "Inner",
3780 "fields": [
3781 {"name": "x", "type": "int"},
3782 {"name": "y", "type": "string"}
3783 ]
3784 }},
3785 {"name": "value", "type": "double"}
3786 ]
3787 }"#;
3788 let schema = AvroSchema::new(schema_json.to_string());
3789 let projected = schema.project(&[1]).unwrap();
3790 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3791 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3792 assert_eq!(fields.len(), 1);
3793 assert_eq!(
3794 fields[0].get("name").and_then(|n| n.as_str()),
3795 Some("inner")
3796 );
3797 let inner_type = fields[0].get("type").unwrap();
3799 assert_eq!(
3800 inner_type.get("type").and_then(|t| t.as_str()),
3801 Some("record")
3802 );
3803 assert_eq!(
3804 inner_type.get("name").and_then(|n| n.as_str()),
3805 Some("Inner")
3806 );
3807 }
3808
3809 #[test]
3810 fn test_project_with_complex_field_types() {
3811 let schema_json = r#"{
3812 "type": "record",
3813 "name": "Test",
3814 "fields": [
3815 {"name": "arr", "type": {"type": "array", "items": "int"}},
3816 {"name": "map", "type": {"type": "map", "values": "string"}},
3817 {"name": "union", "type": ["null", "int"]}
3818 ]
3819 }"#;
3820 let schema = AvroSchema::new(schema_json.to_string());
3821 let projected = schema.project(&[0, 2]).unwrap();
3822 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
3823 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
3824 assert_eq!(fields.len(), 2);
3825 let arr_type = fields[0].get("type").unwrap();
3827 assert_eq!(arr_type.get("type").and_then(|t| t.as_str()), Some("array"));
3828 let union_type = fields[1].get("type").unwrap();
3830 assert!(union_type.is_array());
3831 }
3832
3833 #[test]
3834 fn test_project_error_invalid_json() {
3835 let schema = AvroSchema::new("not valid json".to_string());
3836 let err = schema.project(&[0]).unwrap_err();
3837 let msg = err.to_string();
3838 assert!(
3839 msg.contains("Invalid Avro schema JSON"),
3840 "Expected parse error, got: {msg}"
3841 );
3842 }
3843
3844 #[test]
3845 fn test_project_error_not_object() {
3846 let schema = AvroSchema::new(r#""string""#.to_string());
3848 let err = schema.project(&[0]).unwrap_err();
3849 let msg = err.to_string();
3850 assert!(
3851 msg.contains("must be a JSON object"),
3852 "Expected object error, got: {msg}"
3853 );
3854 }
3855
3856 #[test]
3857 fn test_project_error_array_schema() {
3858 let schema = AvroSchema::new(r#"["null", "int"]"#.to_string());
3860 let err = schema.project(&[0]).unwrap_err();
3861 let msg = err.to_string();
3862 assert!(
3863 msg.contains("must be a JSON object"),
3864 "Expected object error for array schema, got: {msg}"
3865 );
3866 }
3867
3868 #[test]
3869 fn test_project_error_type_not_record() {
3870 let schema_json = r#"{
3871 "type": "enum",
3872 "name": "Color",
3873 "symbols": ["RED", "GREEN", "BLUE"]
3874 }"#;
3875 let schema = AvroSchema::new(schema_json.to_string());
3876 let err = schema.project(&[0]).unwrap_err();
3877 let msg = err.to_string();
3878 assert!(
3879 msg.contains("must be an Avro record") && msg.contains("'enum'"),
3880 "Expected type mismatch error, got: {msg}"
3881 );
3882 }
3883
3884 #[test]
3885 fn test_project_error_type_array() {
3886 let schema_json = r#"{
3887 "type": "array",
3888 "items": "int"
3889 }"#;
3890 let schema = AvroSchema::new(schema_json.to_string());
3891 let err = schema.project(&[0]).unwrap_err();
3892 let msg = err.to_string();
3893 assert!(
3894 msg.contains("must be an Avro record") && msg.contains("'array'"),
3895 "Expected type mismatch error for array type, got: {msg}"
3896 );
3897 }
3898
3899 #[test]
3900 fn test_project_error_type_fixed() {
3901 let schema_json = r#"{
3902 "type": "fixed",
3903 "name": "MD5",
3904 "size": 16
3905 }"#;
3906 let schema = AvroSchema::new(schema_json.to_string());
3907 let err = schema.project(&[0]).unwrap_err();
3908 let msg = err.to_string();
3909 assert!(
3910 msg.contains("must be an Avro record") && msg.contains("'fixed'"),
3911 "Expected type mismatch error for fixed type, got: {msg}"
3912 );
3913 }
3914
3915 #[test]
3916 fn test_project_error_type_map() {
3917 let schema_json = r#"{
3918 "type": "map",
3919 "values": "string"
3920 }"#;
3921 let schema = AvroSchema::new(schema_json.to_string());
3922 let err = schema.project(&[0]).unwrap_err();
3923 let msg = err.to_string();
3924 assert!(
3925 msg.contains("must be an Avro record") && msg.contains("'map'"),
3926 "Expected type mismatch error for map type, got: {msg}"
3927 );
3928 }
3929
3930 #[test]
3931 fn test_project_error_missing_type_field() {
3932 let schema_json = r#"{
3933 "name": "Test",
3934 "fields": [{"name": "a", "type": "int"}]
3935 }"#;
3936 let schema = AvroSchema::new(schema_json.to_string());
3937 let err = schema.project(&[0]).unwrap_err();
3938 let msg = err.to_string();
3939 assert!(
3940 msg.contains("missing required 'type' field"),
3941 "Expected missing type error, got: {msg}"
3942 );
3943 }
3944
3945 #[test]
3946 fn test_project_error_missing_fields() {
3947 let schema_json = r#"{
3948 "type": "record",
3949 "name": "Test"
3950 }"#;
3951 let schema = AvroSchema::new(schema_json.to_string());
3952 let err = schema.project(&[0]).unwrap_err();
3953 let msg = err.to_string();
3954 assert!(
3955 msg.contains("missing required 'fields'"),
3956 "Expected missing fields error, got: {msg}"
3957 );
3958 }
3959
3960 #[test]
3961 fn test_project_error_fields_not_array() {
3962 let schema_json = r#"{
3963 "type": "record",
3964 "name": "Test",
3965 "fields": "not an array"
3966 }"#;
3967 let schema = AvroSchema::new(schema_json.to_string());
3968 let err = schema.project(&[0]).unwrap_err();
3969 let msg = err.to_string();
3970 assert!(
3971 msg.contains("'fields' must be an array"),
3972 "Expected fields array error, got: {msg}"
3973 );
3974 }
3975
3976 #[test]
3977 fn test_project_error_index_out_of_bounds() {
3978 let schema_json = r#"{
3979 "type": "record",
3980 "name": "Test",
3981 "fields": [
3982 {"name": "a", "type": "int"},
3983 {"name": "b", "type": "string"}
3984 ]
3985 }"#;
3986 let schema = AvroSchema::new(schema_json.to_string());
3987 let err = schema.project(&[5]).unwrap_err();
3988 let msg = err.to_string();
3989 assert!(
3990 msg.contains("out of bounds") && msg.contains("5") && msg.contains("2"),
3991 "Expected out of bounds error, got: {msg}"
3992 );
3993 }
3994
3995 #[test]
3996 fn test_project_error_index_out_of_bounds_edge() {
3997 let schema_json = r#"{
3998 "type": "record",
3999 "name": "Test",
4000 "fields": [
4001 {"name": "a", "type": "int"}
4002 ]
4003 }"#;
4004 let schema = AvroSchema::new(schema_json.to_string());
4005 let err = schema.project(&[1]).unwrap_err();
4007 let msg = err.to_string();
4008 assert!(
4009 msg.contains("out of bounds") && msg.contains("1"),
4010 "Expected out of bounds error for edge case, got: {msg}"
4011 );
4012 }
4013
4014 #[test]
4015 fn test_project_error_duplicate_index() {
4016 let schema_json = r#"{
4017 "type": "record",
4018 "name": "Test",
4019 "fields": [
4020 {"name": "a", "type": "int"},
4021 {"name": "b", "type": "string"},
4022 {"name": "c", "type": "long"}
4023 ]
4024 }"#;
4025 let schema = AvroSchema::new(schema_json.to_string());
4026 let err = schema.project(&[0, 1, 0]).unwrap_err();
4027 let msg = err.to_string();
4028 assert!(
4029 msg.contains("Duplicate projection index") && msg.contains("0"),
4030 "Expected duplicate index error, got: {msg}"
4031 );
4032 }
4033
4034 #[test]
4035 fn test_project_error_duplicate_index_consecutive() {
4036 let schema_json = r#"{
4037 "type": "record",
4038 "name": "Test",
4039 "fields": [
4040 {"name": "a", "type": "int"},
4041 {"name": "b", "type": "string"}
4042 ]
4043 }"#;
4044 let schema = AvroSchema::new(schema_json.to_string());
4045 let err = schema.project(&[1, 1]).unwrap_err();
4046 let msg = err.to_string();
4047 assert!(
4048 msg.contains("Duplicate projection index") && msg.contains("1"),
4049 "Expected duplicate index error for consecutive duplicates, got: {msg}"
4050 );
4051 }
4052
4053 #[test]
4054 fn test_project_with_empty_fields() {
4055 let schema_json = r#"{
4056 "type": "record",
4057 "name": "EmptyRecord",
4058 "fields": []
4059 }"#;
4060 let schema = AvroSchema::new(schema_json.to_string());
4061 let projected = schema.project(&[]).unwrap();
4063 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
4064 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
4065 assert!(fields.is_empty());
4066 }
4067
4068 #[test]
4069 fn test_project_empty_fields_index_out_of_bounds() {
4070 let schema_json = r#"{
4071 "type": "record",
4072 "name": "EmptyRecord",
4073 "fields": []
4074 }"#;
4075 let schema = AvroSchema::new(schema_json.to_string());
4076 let err = schema.project(&[0]).unwrap_err();
4077 let msg = err.to_string();
4078 assert!(
4079 msg.contains("out of bounds") && msg.contains("0 fields"),
4080 "Expected out of bounds error for empty record, got: {msg}"
4081 );
4082 }
4083
4084 #[test]
4085 fn test_project_result_is_valid_avro_schema() {
4086 let schema_json = r#"{
4087 "type": "record",
4088 "name": "Test",
4089 "namespace": "com.example",
4090 "fields": [
4091 {"name": "id", "type": "long"},
4092 {"name": "name", "type": "string"},
4093 {"name": "active", "type": "boolean"}
4094 ]
4095 }"#;
4096 let schema = AvroSchema::new(schema_json.to_string());
4097 let projected = schema.project(&[0, 2]).unwrap();
4098 let parsed = projected.schema();
4100 assert!(parsed.is_ok(), "Projected schema should be valid Avro");
4101 match parsed.unwrap() {
4102 Schema::Complex(ComplexType::Record(r)) => {
4103 assert_eq!(r.name, "Test");
4104 assert_eq!(r.namespace, Some("com.example"));
4105 assert_eq!(r.fields.len(), 2);
4106 assert_eq!(r.fields[0].name, "id");
4107 assert_eq!(r.fields[1].name, "active");
4108 }
4109 _ => panic!("Expected Record schema"),
4110 }
4111 }
4112
4113 #[test]
4114 fn test_project_non_contiguous_indices() {
4115 let schema_json = r#"{
4116 "type": "record",
4117 "name": "Test",
4118 "fields": [
4119 {"name": "f0", "type": "int"},
4120 {"name": "f1", "type": "int"},
4121 {"name": "f2", "type": "int"},
4122 {"name": "f3", "type": "int"},
4123 {"name": "f4", "type": "int"}
4124 ]
4125 }"#;
4126 let schema = AvroSchema::new(schema_json.to_string());
4127 let projected = schema.project(&[0, 2, 4]).unwrap();
4129 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
4130 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
4131 assert_eq!(fields.len(), 3);
4132 assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("f0"));
4133 assert_eq!(fields[1].get("name").and_then(|n| n.as_str()), Some("f2"));
4134 assert_eq!(fields[2].get("name").and_then(|n| n.as_str()), Some("f4"));
4135 }
4136
4137 #[test]
4138 fn test_project_single_field_from_many() {
4139 let schema_json = r#"{
4140 "type": "record",
4141 "name": "BigRecord",
4142 "fields": [
4143 {"name": "f0", "type": "int"},
4144 {"name": "f1", "type": "int"},
4145 {"name": "f2", "type": "int"},
4146 {"name": "f3", "type": "int"},
4147 {"name": "f4", "type": "int"},
4148 {"name": "f5", "type": "int"},
4149 {"name": "f6", "type": "int"},
4150 {"name": "f7", "type": "int"},
4151 {"name": "f8", "type": "int"},
4152 {"name": "f9", "type": "int"}
4153 ]
4154 }"#;
4155 let schema = AvroSchema::new(schema_json.to_string());
4156 let projected = schema.project(&[9]).unwrap();
4158 let v: Value = serde_json::from_str(&projected.json_string).unwrap();
4159 let fields = v.get("fields").and_then(|f| f.as_array()).unwrap();
4160 assert_eq!(fields.len(), 1);
4161 assert_eq!(fields[0].get("name").and_then(|n| n.as_str()), Some("f9"));
4162 }
4163}