1use crate::codec::{
21 AvroDataType, AvroLiteral, Codec, EnumMapping, Promotion, ResolutionInfo, ResolvedField,
22 ResolvedRecord, ResolvedUnion, Tz,
23};
24use crate::errors::AvroError;
25use crate::reader::cursor::AvroCursor;
26use crate::schema::Nullability;
27#[cfg(feature = "small_decimals")]
28use arrow_array::builder::{Decimal32Builder, Decimal64Builder};
29use arrow_array::builder::{Decimal128Builder, Decimal256Builder, IntervalMonthDayNanoBuilder};
30use arrow_array::types::*;
31use arrow_array::*;
32use arrow_buffer::*;
33#[cfg(feature = "small_decimals")]
34use arrow_schema::{DECIMAL32_MAX_PRECISION, DECIMAL64_MAX_PRECISION};
35use arrow_schema::{
36 DECIMAL128_MAX_PRECISION, DECIMAL256_MAX_PRECISION, DataType, Field as ArrowField, FieldRef,
37 Fields, Schema as ArrowSchema, SchemaRef, UnionFields, UnionMode,
38};
39#[cfg(feature = "avro_custom_types")]
40use arrow_select::take::{TakeOptions, take};
41use strum_macros::AsRefStr;
42use uuid::Uuid;
43
44use std::cmp::Ordering;
45use std::mem;
46use std::sync::Arc;
47
48const DEFAULT_CAPACITY: usize = 1024;
49
50macro_rules! decode_decimal {
52 ($size:expr, $buf:expr, $builder:expr, $N:expr, $Int:ty) => {{
53 let bytes = read_decimal_bytes_be::<{ $N }>($buf, *$size)?;
54 $builder.append_value(<$Int>::from_be_bytes(bytes));
55 }};
56}
57
58macro_rules! flush_decimal {
60 ($builder:expr, $precision:expr, $scale:expr, $nulls:expr, $ArrayTy:ty) => {{
61 let (_, vals, _) = $builder.finish().into_parts();
62 let dec = <$ArrayTy>::try_new(vals, $nulls)?
63 .with_precision_and_scale(*$precision as u8, $scale.unwrap_or(0) as i8)?;
64 Arc::new(dec) as ArrayRef
65 }};
66}
67
68macro_rules! append_decimal_default {
71 ($lit:expr, $builder:expr, $N:literal, $Int:ty, $name:literal) => {{
72 match $lit {
73 AvroLiteral::Bytes(b) => {
74 let ext = sign_cast_to::<$N>(b)?;
75 let val = <$Int>::from_be_bytes(ext);
76 $builder.append_value(val);
77 Ok(())
78 }
79 _ => Err(AvroError::InvalidArgument(
80 concat!(
81 "Default for ",
82 $name,
83 " must be bytes (two's-complement big-endian)"
84 )
85 .to_string(),
86 )),
87 }
88 }};
89}
90
91#[derive(Debug)]
93pub(crate) struct RecordDecoder {
94 schema: SchemaRef,
95 fields: Vec<Decoder>,
96 projector: Option<Projector>,
97 row_count: usize,
98}
99
100impl RecordDecoder {
101 pub(crate) fn try_new_with_options(data_type: &AvroDataType) -> Result<Self, AvroError> {
112 match data_type.codec() {
113 Codec::Struct(reader_fields) => {
114 let mut arrow_fields = Vec::with_capacity(reader_fields.len());
116 let mut encodings = Vec::with_capacity(reader_fields.len());
117 let mut field_defaults = Vec::with_capacity(reader_fields.len());
118 for avro_field in reader_fields.iter() {
119 arrow_fields.push(avro_field.field());
120 encodings.push(Decoder::try_new(avro_field.data_type())?);
121
122 if let Some(ResolutionInfo::DefaultValue(lit)) =
123 avro_field.data_type().resolution.as_ref()
124 {
125 field_defaults.push(Some(lit.clone()));
126 } else {
127 field_defaults.push(None);
128 }
129 }
130 let projector = match data_type.resolution.as_ref() {
131 Some(ResolutionInfo::Record(rec)) => {
132 Some(ProjectorBuilder::try_new(rec, &field_defaults).build()?)
133 }
134 _ => None,
135 };
136 Ok(Self {
137 schema: Arc::new(ArrowSchema::new(arrow_fields)),
138 fields: encodings,
139 projector,
140 row_count: 0,
141 })
142 }
143 other => Err(AvroError::ParseError(format!(
144 "Expected record got {other:?}"
145 ))),
146 }
147 }
148
149 pub(crate) fn schema(&self) -> &SchemaRef {
151 &self.schema
152 }
153
154 pub(crate) fn decode(&mut self, buf: &[u8], count: usize) -> Result<usize, AvroError> {
156 let mut cursor = AvroCursor::new(buf);
157 match self.projector.as_mut() {
158 Some(proj) => {
159 for _ in 0..count {
160 proj.project_record(&mut cursor, &mut self.fields)?;
161 }
162 }
163 None => {
164 for _ in 0..count {
165 for field in &mut self.fields {
166 field.decode(&mut cursor)?;
167 }
168 }
169 }
170 }
171 self.row_count += count;
172 Ok(cursor.position())
173 }
174
175 pub(crate) fn flush(&mut self) -> Result<RecordBatch, AvroError> {
177 let arrays = self
178 .fields
179 .iter_mut()
180 .map(|x| x.flush(None))
181 .collect::<Result<Vec<_>, _>>()?;
182 let batch_options = RecordBatchOptions::new().with_row_count(Some(self.row_count));
183 self.row_count = 0;
184 RecordBatch::try_new_with_options(self.schema.clone(), arrays, &batch_options)
185 .map_err(Into::into)
186 }
187}
188
189#[derive(Debug, AsRefStr)]
190enum Decoder {
191 Null(usize),
192 Boolean(BooleanBufferBuilder),
193 Int32(Vec<i32>),
194 Int64(Vec<i64>),
195 #[cfg(feature = "avro_custom_types")]
196 DurationSecond(Vec<i64>),
197 #[cfg(feature = "avro_custom_types")]
198 DurationMillisecond(Vec<i64>),
199 #[cfg(feature = "avro_custom_types")]
200 DurationMicrosecond(Vec<i64>),
201 #[cfg(feature = "avro_custom_types")]
202 DurationNanosecond(Vec<i64>),
203 #[cfg(feature = "avro_custom_types")]
204 Int8(Vec<i8>),
205 #[cfg(feature = "avro_custom_types")]
206 Int16(Vec<i16>),
207 #[cfg(feature = "avro_custom_types")]
208 UInt8(Vec<u8>),
209 #[cfg(feature = "avro_custom_types")]
210 UInt16(Vec<u16>),
211 #[cfg(feature = "avro_custom_types")]
212 UInt32(Vec<u32>),
213 #[cfg(feature = "avro_custom_types")]
214 UInt64(Vec<u64>),
215 #[cfg(feature = "avro_custom_types")]
216 Float16(Vec<u16>), #[cfg(feature = "avro_custom_types")]
218 Date64(Vec<i64>),
219 #[cfg(feature = "avro_custom_types")]
220 TimeNanos(Vec<i64>),
221 #[cfg(feature = "avro_custom_types")]
222 Time32Secs(Vec<i32>),
223 #[cfg(feature = "avro_custom_types")]
224 TimestampSecs(bool, Vec<i64>),
225 #[cfg(feature = "avro_custom_types")]
226 IntervalYearMonth(Vec<i32>),
227 #[cfg(feature = "avro_custom_types")]
228 IntervalMonthDayNano(Vec<IntervalMonthDayNano>),
229 #[cfg(feature = "avro_custom_types")]
230 IntervalDayTime(Vec<IntervalDayTime>),
231 Float32(Vec<f32>),
232 Float64(Vec<f64>),
233 Date32(Vec<i32>),
234 TimeMillis(Vec<i32>),
235 TimeMicros(Vec<i64>),
236 TimestampMillis(Option<Tz>, Vec<i64>),
237 TimestampMicros(Option<Tz>, Vec<i64>),
238 TimestampNanos(Option<Tz>, Vec<i64>),
239 Int32ToInt64(Vec<i64>),
240 Int32ToFloat32(Vec<f32>),
241 Int32ToFloat64(Vec<f64>),
242 Int64ToFloat32(Vec<f32>),
243 Int64ToFloat64(Vec<f64>),
244 Float32ToFloat64(Vec<f64>),
245 BytesToString(OffsetBufferBuilder<i32>, Vec<u8>),
246 StringToBytes(OffsetBufferBuilder<i32>, Vec<u8>),
247 Binary(OffsetBufferBuilder<i32>, Vec<u8>),
248 String(OffsetBufferBuilder<i32>, Vec<u8>),
250 StringView(OffsetBufferBuilder<i32>, Vec<u8>),
252 Array(FieldRef, OffsetBufferBuilder<i32>, Box<Decoder>),
253 Record {
254 fields: Fields,
255 decoders: Vec<Decoder>,
256 defaults: Vec<Option<AvroLiteral>>,
257 projector: Option<Projector>,
258 row_count: usize,
262 },
263 Map(
264 FieldRef,
265 OffsetBufferBuilder<i32>,
266 OffsetBufferBuilder<i32>,
267 Vec<u8>,
268 Box<Decoder>,
269 ),
270 Fixed(i32, Vec<u8>),
271 Enum(Vec<i32>, Arc<[String]>, Option<EnumResolution>),
272 Duration(IntervalMonthDayNanoBuilder),
273 Uuid(Vec<u8>),
274 #[cfg(feature = "small_decimals")]
275 Decimal32(usize, Option<usize>, Option<usize>, Decimal32Builder),
276 #[cfg(feature = "small_decimals")]
277 Decimal64(usize, Option<usize>, Option<usize>, Decimal64Builder),
278 Decimal128(usize, Option<usize>, Option<usize>, Decimal128Builder),
279 Decimal256(usize, Option<usize>, Option<usize>, Decimal256Builder),
280 #[cfg(feature = "avro_custom_types")]
281 RunEndEncoded(u8, usize, Box<Decoder>),
282 Union(UnionDecoder),
283 Nullable(NullableDecoder),
285}
286
287#[derive(Debug)]
288struct NullableDecoder {
289 plan: NullablePlan,
290 validity: NullBufferBuilder,
291 values: Box<Decoder>,
292 pending_nulls: usize,
293}
294
295impl NullableDecoder {
296 fn new(plan: NullablePlan, values: Decoder) -> Self {
297 Self {
298 plan,
299 validity: NullBufferBuilder::new(DEFAULT_CAPACITY),
300 values: Box::new(values),
301 pending_nulls: 0,
302 }
303 }
304
305 #[inline]
306 fn materialize_pending(&mut self) -> Result<(), AvroError> {
307 if self.pending_nulls == 0 {
308 return Ok(());
309 }
310
311 self.values.append_nulls(self.pending_nulls)?;
312 self.validity.append_n_nulls(self.pending_nulls);
313 self.pending_nulls = 0;
314 Ok(())
315 }
316}
317
318impl Decoder {
319 fn try_new(data_type: &AvroDataType) -> Result<Self, AvroError> {
320 if let Some(ResolutionInfo::Union(info)) = data_type.resolution.as_ref()
321 && info.writer_is_union
322 && !info.reader_is_union
323 {
324 let mut clone = data_type.clone();
325 clone.resolution = None; let target = Self::try_new_internal(&clone)?;
327 let decoder = Self::Union(
328 UnionDecoderBuilder::new()
329 .with_resolved_union(info.clone())
330 .with_target(target)
331 .build()?,
332 );
333 return Ok(decoder);
334 }
335 Self::try_new_internal(data_type)
336 }
337
338 fn try_new_internal(data_type: &AvroDataType) -> Result<Self, AvroError> {
339 let promotion = match data_type.resolution.as_ref() {
341 Some(ResolutionInfo::Promotion(p)) => Some(*p),
342 _ => None,
343 };
344 let decoder = match (data_type.codec(), promotion) {
345 (Codec::Int64, Some(Promotion::IntToLong)) => {
346 Self::Int32ToInt64(Vec::with_capacity(DEFAULT_CAPACITY))
347 }
348 (Codec::Float32, Some(Promotion::IntToFloat)) => {
349 Self::Int32ToFloat32(Vec::with_capacity(DEFAULT_CAPACITY))
350 }
351 (Codec::Float64, Some(Promotion::IntToDouble)) => {
352 Self::Int32ToFloat64(Vec::with_capacity(DEFAULT_CAPACITY))
353 }
354 (Codec::Float32, Some(Promotion::LongToFloat)) => {
355 Self::Int64ToFloat32(Vec::with_capacity(DEFAULT_CAPACITY))
356 }
357 (Codec::Float64, Some(Promotion::LongToDouble)) => {
358 Self::Int64ToFloat64(Vec::with_capacity(DEFAULT_CAPACITY))
359 }
360 (Codec::Float64, Some(Promotion::FloatToDouble)) => {
361 Self::Float32ToFloat64(Vec::with_capacity(DEFAULT_CAPACITY))
362 }
363 (Codec::Utf8 | Codec::Utf8View, Some(Promotion::BytesToString)) => Self::BytesToString(
364 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
365 Vec::with_capacity(DEFAULT_CAPACITY),
366 ),
367 (Codec::Binary, Some(Promotion::StringToBytes)) => Self::StringToBytes(
368 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
369 Vec::with_capacity(DEFAULT_CAPACITY),
370 ),
371 (Codec::Null, _) => Self::Null(0),
372 (Codec::Boolean, _) => Self::Boolean(BooleanBufferBuilder::new(DEFAULT_CAPACITY)),
373 (Codec::Int32, _) => Self::Int32(Vec::with_capacity(DEFAULT_CAPACITY)),
374 (Codec::Int64, _) => Self::Int64(Vec::with_capacity(DEFAULT_CAPACITY)),
375 (Codec::Float32, _) => Self::Float32(Vec::with_capacity(DEFAULT_CAPACITY)),
376 (Codec::Float64, _) => Self::Float64(Vec::with_capacity(DEFAULT_CAPACITY)),
377 (Codec::Binary, _) => Self::Binary(
378 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
379 Vec::with_capacity(DEFAULT_CAPACITY),
380 ),
381 (Codec::Utf8, _) => Self::String(
382 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
383 Vec::with_capacity(DEFAULT_CAPACITY),
384 ),
385 (Codec::Utf8View, _) => Self::StringView(
386 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
387 Vec::with_capacity(DEFAULT_CAPACITY),
388 ),
389 (Codec::Date32, _) => Self::Date32(Vec::with_capacity(DEFAULT_CAPACITY)),
390 (Codec::TimeMillis, _) => Self::TimeMillis(Vec::with_capacity(DEFAULT_CAPACITY)),
391 (Codec::TimeMicros, _) => Self::TimeMicros(Vec::with_capacity(DEFAULT_CAPACITY)),
392 (Codec::TimestampMillis(tz), _) => {
393 Self::TimestampMillis(*tz, Vec::with_capacity(DEFAULT_CAPACITY))
394 }
395 (Codec::TimestampMicros(tz), _) => {
396 Self::TimestampMicros(*tz, Vec::with_capacity(DEFAULT_CAPACITY))
397 }
398 (Codec::TimestampNanos(tz), _) => {
399 Self::TimestampNanos(*tz, Vec::with_capacity(DEFAULT_CAPACITY))
400 }
401 #[cfg(feature = "avro_custom_types")]
402 (Codec::DurationNanos, _) => {
403 Self::DurationNanosecond(Vec::with_capacity(DEFAULT_CAPACITY))
404 }
405 #[cfg(feature = "avro_custom_types")]
406 (Codec::DurationMicros, _) => {
407 Self::DurationMicrosecond(Vec::with_capacity(DEFAULT_CAPACITY))
408 }
409 #[cfg(feature = "avro_custom_types")]
410 (Codec::DurationMillis, _) => {
411 Self::DurationMillisecond(Vec::with_capacity(DEFAULT_CAPACITY))
412 }
413 #[cfg(feature = "avro_custom_types")]
414 (Codec::DurationSeconds, _) => {
415 Self::DurationSecond(Vec::with_capacity(DEFAULT_CAPACITY))
416 }
417 #[cfg(feature = "avro_custom_types")]
418 (Codec::Int8, _) => Self::Int8(Vec::with_capacity(DEFAULT_CAPACITY)),
419 #[cfg(feature = "avro_custom_types")]
420 (Codec::Int16, _) => Self::Int16(Vec::with_capacity(DEFAULT_CAPACITY)),
421 #[cfg(feature = "avro_custom_types")]
422 (Codec::UInt8, _) => Self::UInt8(Vec::with_capacity(DEFAULT_CAPACITY)),
423 #[cfg(feature = "avro_custom_types")]
424 (Codec::UInt16, _) => Self::UInt16(Vec::with_capacity(DEFAULT_CAPACITY)),
425 #[cfg(feature = "avro_custom_types")]
426 (Codec::UInt32, _) => Self::UInt32(Vec::with_capacity(DEFAULT_CAPACITY)),
427 #[cfg(feature = "avro_custom_types")]
428 (Codec::UInt64, _) => Self::UInt64(Vec::with_capacity(DEFAULT_CAPACITY)),
429 #[cfg(feature = "avro_custom_types")]
430 (Codec::Float16, _) => Self::Float16(Vec::with_capacity(DEFAULT_CAPACITY)),
431 #[cfg(feature = "avro_custom_types")]
432 (Codec::Date64, _) => Self::Date64(Vec::with_capacity(DEFAULT_CAPACITY)),
433 #[cfg(feature = "avro_custom_types")]
434 (Codec::TimeNanos, _) => Self::TimeNanos(Vec::with_capacity(DEFAULT_CAPACITY)),
435 #[cfg(feature = "avro_custom_types")]
436 (Codec::Time32Secs, _) => Self::Time32Secs(Vec::with_capacity(DEFAULT_CAPACITY)),
437 #[cfg(feature = "avro_custom_types")]
438 (Codec::TimestampSecs(is_utc), _) => {
439 Self::TimestampSecs(*is_utc, Vec::with_capacity(DEFAULT_CAPACITY))
440 }
441 #[cfg(feature = "avro_custom_types")]
442 (Codec::IntervalYearMonth, _) => {
443 Self::IntervalYearMonth(Vec::with_capacity(DEFAULT_CAPACITY))
444 }
445 #[cfg(feature = "avro_custom_types")]
446 (Codec::IntervalMonthDayNano, _) => {
447 Self::IntervalMonthDayNano(Vec::with_capacity(DEFAULT_CAPACITY))
448 }
449 #[cfg(feature = "avro_custom_types")]
450 (Codec::IntervalDayTime, _) => {
451 Self::IntervalDayTime(Vec::with_capacity(DEFAULT_CAPACITY))
452 }
453 (Codec::Fixed(sz), _) => Self::Fixed(*sz, Vec::with_capacity(DEFAULT_CAPACITY)),
454 (Codec::Decimal(precision, scale, size), _) => {
455 let p = *precision;
456 let s = *scale;
457 let prec = p as u8;
458 let scl = s.unwrap_or(0) as i8;
459 #[cfg(feature = "small_decimals")]
460 {
461 if p <= DECIMAL32_MAX_PRECISION as usize {
462 let builder = Decimal32Builder::with_capacity(DEFAULT_CAPACITY)
463 .with_precision_and_scale(prec, scl)?;
464 Self::Decimal32(p, s, *size, builder)
465 } else if p <= DECIMAL64_MAX_PRECISION as usize {
466 let builder = Decimal64Builder::with_capacity(DEFAULT_CAPACITY)
467 .with_precision_and_scale(prec, scl)?;
468 Self::Decimal64(p, s, *size, builder)
469 } else if p <= DECIMAL128_MAX_PRECISION as usize {
470 let builder = Decimal128Builder::with_capacity(DEFAULT_CAPACITY)
471 .with_precision_and_scale(prec, scl)?;
472 Self::Decimal128(p, s, *size, builder)
473 } else if p <= DECIMAL256_MAX_PRECISION as usize {
474 let builder = Decimal256Builder::with_capacity(DEFAULT_CAPACITY)
475 .with_precision_and_scale(prec, scl)?;
476 Self::Decimal256(p, s, *size, builder)
477 } else {
478 return Err(AvroError::ParseError(format!(
479 "Decimal precision {p} exceeds maximum supported"
480 )));
481 }
482 }
483 #[cfg(not(feature = "small_decimals"))]
484 {
485 if p <= DECIMAL128_MAX_PRECISION as usize {
486 let builder = Decimal128Builder::with_capacity(DEFAULT_CAPACITY)
487 .with_precision_and_scale(prec, scl)?;
488 Self::Decimal128(p, s, *size, builder)
489 } else if p <= DECIMAL256_MAX_PRECISION as usize {
490 let builder = Decimal256Builder::with_capacity(DEFAULT_CAPACITY)
491 .with_precision_and_scale(prec, scl)?;
492 Self::Decimal256(p, s, *size, builder)
493 } else {
494 return Err(AvroError::ParseError(format!(
495 "Decimal precision {p} exceeds maximum supported"
496 )));
497 }
498 }
499 }
500 (Codec::Interval, _) => Self::Duration(IntervalMonthDayNanoBuilder::new()),
501 (Codec::List(item), _) => {
502 let decoder = Self::try_new(item)?;
503 Self::Array(
504 Arc::new(item.field_with_name("item")),
505 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
506 Box::new(decoder),
507 )
508 }
509 (Codec::Enum(symbols), _) => {
510 let res = match data_type.resolution.as_ref() {
511 Some(ResolutionInfo::EnumMapping(mapping)) => {
512 Some(EnumResolution::new(mapping))
513 }
514 _ => None,
515 };
516 Self::Enum(Vec::with_capacity(DEFAULT_CAPACITY), symbols.clone(), res)
517 }
518 (Codec::Struct(fields), _) => {
519 let mut arrow_fields = Vec::with_capacity(fields.len());
520 let mut encodings = Vec::with_capacity(fields.len());
521 let mut field_defaults = Vec::with_capacity(fields.len());
522 for avro_field in fields.iter() {
523 let encoding = Self::try_new(avro_field.data_type())?;
524 arrow_fields.push(avro_field.field());
525 encodings.push(encoding);
526
527 if let Some(ResolutionInfo::DefaultValue(lit)) =
528 avro_field.data_type().resolution.as_ref()
529 {
530 field_defaults.push(Some(lit.clone()));
531 } else {
532 field_defaults.push(None);
533 }
534 }
535 let projector =
536 if let Some(ResolutionInfo::Record(rec)) = data_type.resolution.as_ref() {
537 Some(ProjectorBuilder::try_new(rec, &field_defaults).build()?)
538 } else {
539 None
540 };
541 Self::Record {
542 fields: arrow_fields.into(),
543 decoders: encodings,
544 defaults: field_defaults,
545 projector,
546 row_count: 0,
547 }
548 }
549 (Codec::Map(child), _) => {
550 let val_field = child.field_with_name(ArrowField::MAP_VALUE_FIELD_DEFAULT_NAME);
551 let map_field = Arc::new(ArrowField::new(
552 ArrowField::MAP_ENTRIES_FIELD_DEFAULT_NAME,
553 DataType::Struct(Fields::from(vec![
554 ArrowField::new(
555 ArrowField::MAP_KEY_FIELD_DEFAULT_NAME,
556 DataType::Utf8,
557 false,
558 ),
559 val_field,
560 ])),
561 false,
562 ));
563 let val_dec = Self::try_new(child)?;
564 Self::Map(
565 map_field,
566 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
567 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
568 Vec::with_capacity(DEFAULT_CAPACITY),
569 Box::new(val_dec),
570 )
571 }
572 (Codec::Uuid, _) => Self::Uuid(Vec::with_capacity(DEFAULT_CAPACITY)),
573 (Codec::Union(encodings, fields, UnionMode::Dense), _) => {
574 let decoders = encodings
575 .iter()
576 .map(Self::try_new_internal)
577 .collect::<Result<Vec<_>, _>>()?;
578 if fields.len() != decoders.len() {
579 return Err(AvroError::SchemaError(format!(
580 "Union has {} fields but {} decoders",
581 fields.len(),
582 decoders.len()
583 )));
584 }
585 let branch_count = decoders.len();
588 let max_addr = (i32::MAX as usize) + 1;
589 if branch_count > max_addr {
590 return Err(AvroError::SchemaError(format!(
591 "Union has {branch_count} branches, which exceeds the maximum addressable \
592 branches by an Avro int tag ({} + 1).",
593 i32::MAX
594 )));
595 }
596 let mut builder = UnionDecoderBuilder::new()
597 .with_fields(fields.clone())
598 .with_branches(decoders);
599 if let Some(ResolutionInfo::Union(info)) = data_type.resolution.as_ref()
600 && info.reader_is_union
601 {
602 builder = builder.with_resolved_union(info.clone());
603 }
604 Self::Union(builder.build()?)
605 }
606 (Codec::Union(_, _, _), _) => {
607 return Err(AvroError::NYI(
608 "Sparse Arrow unions are not yet supported".to_string(),
609 ));
610 }
611 #[cfg(feature = "avro_custom_types")]
612 (Codec::RunEndEncoded(values_dt, width_bits_or_bytes), _) => {
613 let inner = Self::try_new(values_dt)?;
614 let byte_width: u8 = match *width_bits_or_bytes {
615 2 | 4 | 8 => *width_bits_or_bytes,
616 16 => 2,
617 32 => 4,
618 64 => 8,
619 other => {
620 return Err(AvroError::InvalidArgument(format!(
621 "Unsupported run-end width {other} for RunEndEncoded; \
622 expected 16/32/64 bits or 2/4/8 bytes"
623 )));
624 }
625 };
626 Self::RunEndEncoded(byte_width, 0, Box::new(inner))
627 }
628 };
629 Ok(match data_type.nullability() {
630 Some(nullability) => {
631 let plan = match &data_type.resolution {
633 None => NullablePlan::ReadTag {
634 nullability,
635 resolution: ResolutionPlan::Promotion(Promotion::Direct),
636 },
637 Some(ResolutionInfo::Promotion(_)) => {
638 NullablePlan::FromSingle {
641 resolution: ResolutionPlan::Promotion(Promotion::Direct),
642 }
643 }
644 Some(ResolutionInfo::Union(info)) if !info.writer_is_union => {
645 let Some(Some((_, resolution))) = info.writer_to_reader.first() else {
646 return Err(AvroError::SchemaError(
647 "unexpected union resolution info for non-union writer and union reader type".into(),
648 ));
649 };
650 let resolution = ResolutionPlan::try_new(&decoder, resolution)?;
651 NullablePlan::FromSingle { resolution }
652 }
653 Some(ResolutionInfo::Union(info)) => {
654 let Some((_, resolution)) =
655 info.writer_to_reader[nullability.non_null_index()].as_ref()
656 else {
657 return Err(AvroError::SchemaError(
658 "unexpected union resolution info for nullable writer type".into(),
659 ));
660 };
661 NullablePlan::ReadTag {
662 nullability,
663 resolution: ResolutionPlan::try_new(&decoder, resolution)?,
664 }
665 }
666 Some(resolution) => NullablePlan::FromSingle {
667 resolution: ResolutionPlan::try_new(&decoder, resolution)?,
668 },
669 };
670 Self::Nullable(NullableDecoder::new(plan, decoder))
671 }
672 None => decoder,
673 })
674 }
675
676 fn append_null(&mut self) -> Result<(), AvroError> {
678 self.append_nulls(1)
679 }
680
681 fn append_nulls(&mut self, count: usize) -> Result<(), AvroError> {
684 if count == 0 {
685 return Ok(());
686 }
687 match self {
688 Self::Null(size) => *size += count,
689 Self::Boolean(values) => values.append_n(count, false),
690 Self::Int32(values) | Self::Date32(values) | Self::TimeMillis(values) => {
691 values.resize(values.len() + count, 0)
692 }
693 Self::Int64(values)
694 | Self::Int32ToInt64(values)
695 | Self::TimeMicros(values)
696 | Self::TimestampMillis(_, values)
697 | Self::TimestampMicros(_, values)
698 | Self::TimestampNanos(_, values) => values.resize(values.len() + count, 0),
699 #[cfg(feature = "avro_custom_types")]
700 Self::DurationSecond(values)
701 | Self::DurationMillisecond(values)
702 | Self::DurationMicrosecond(values)
703 | Self::DurationNanosecond(values)
704 | Self::Date64(values)
705 | Self::TimeNanos(values)
706 | Self::TimestampSecs(_, values) => values.resize(values.len() + count, 0),
707 #[cfg(feature = "avro_custom_types")]
708 Self::Int8(values) => values.resize(values.len() + count, 0),
709 #[cfg(feature = "avro_custom_types")]
710 Self::Int16(values) => values.resize(values.len() + count, 0),
711 #[cfg(feature = "avro_custom_types")]
712 Self::UInt8(values) => values.resize(values.len() + count, 0),
713 #[cfg(feature = "avro_custom_types")]
714 Self::UInt16(values) | Self::Float16(values) => values.resize(values.len() + count, 0),
715 #[cfg(feature = "avro_custom_types")]
716 Self::UInt32(values) => values.resize(values.len() + count, 0),
717 #[cfg(feature = "avro_custom_types")]
718 Self::UInt64(values) => values.resize(values.len() + count, 0),
719 #[cfg(feature = "avro_custom_types")]
720 Self::Time32Secs(values) | Self::IntervalYearMonth(values) => {
721 values.resize(values.len() + count, 0)
722 }
723 #[cfg(feature = "avro_custom_types")]
724 Self::IntervalDayTime(values) => {
725 values.resize(values.len() + count, IntervalDayTime::new(0, 0))
726 }
727 #[cfg(feature = "avro_custom_types")]
728 Self::IntervalMonthDayNano(values) => {
729 values.resize(values.len() + count, IntervalMonthDayNano::new(0, 0, 0))
730 }
731 Self::Float32(values) | Self::Int32ToFloat32(values) | Self::Int64ToFloat32(values) => {
732 values.resize(values.len() + count, 0.0)
733 }
734 Self::Float64(values)
735 | Self::Int32ToFloat64(values)
736 | Self::Int64ToFloat64(values)
737 | Self::Float32ToFloat64(values) => values.resize(values.len() + count, 0.0),
738 Self::Binary(offsets, _)
739 | Self::String(offsets, _)
740 | Self::StringView(offsets, _)
741 | Self::BytesToString(offsets, _)
742 | Self::StringToBytes(offsets, _)
743 | Self::Array(_, offsets, _)
744 | Self::Map(_, _, offsets, _, _) => {
745 offsets.reserve(count);
746 for _ in 0..count {
747 offsets.push_length(0);
748 }
749 }
750 Self::Record {
751 decoders,
752 row_count,
753 ..
754 } => {
755 for child in decoders.iter_mut() {
756 child.append_nulls(count)?;
757 }
758 *row_count += count;
759 }
760 Self::Fixed(width, values) => {
761 values.resize(values.len() + (*width as usize) * count, 0)
762 }
763 Self::Enum(values, _, _) => values.resize(values.len() + count, 0),
764 Self::Duration(builder) => builder.append_nulls(count),
765 Self::Uuid(values) => values.resize(values.len() + 16 * count, 0),
766 #[cfg(feature = "small_decimals")]
767 Self::Decimal32(_, _, _, builder) => builder.append_value_n(0, count),
768 #[cfg(feature = "small_decimals")]
769 Self::Decimal64(_, _, _, builder) => builder.append_value_n(0, count),
770 Self::Decimal128(_, _, _, builder) => builder.append_value_n(0, count),
771 Self::Decimal256(_, _, _, builder) => builder.append_value_n(i256::ZERO, count),
772 #[cfg(feature = "avro_custom_types")]
773 Self::RunEndEncoded(_, len, inner) => {
774 inner.append_nulls(count)?;
775 *len += count;
776 }
777 Self::Union(union) => {
778 for _ in 0..count {
779 union.append_null()?;
780 }
781 }
782 Self::Nullable(nullable) => nullable.pending_nulls += count,
783 }
784 Ok(())
785 }
786
787 fn append_default(&mut self, lit: &AvroLiteral) -> Result<(), AvroError> {
789 match self {
790 Self::Nullable(nullable) => {
791 if matches!(lit, AvroLiteral::Null) {
792 nullable.pending_nulls += 1;
793 } else {
794 nullable.materialize_pending()?;
795 nullable.values.append_default(lit)?;
796 nullable.validity.append_non_null();
797 }
798 Ok(())
799 }
800 Self::Null(count) => match lit {
801 AvroLiteral::Null => {
802 *count += 1;
803 Ok(())
804 }
805 _ => Err(AvroError::InvalidArgument(
806 "Non-null default for null type".to_string(),
807 )),
808 },
809 Self::Boolean(b) => match lit {
810 AvroLiteral::Boolean(v) => {
811 b.append(*v);
812 Ok(())
813 }
814 _ => Err(AvroError::InvalidArgument(
815 "Default for boolean must be boolean".to_string(),
816 )),
817 },
818 Self::Int32(v) | Self::Date32(v) | Self::TimeMillis(v) => match lit {
819 AvroLiteral::Int(i) => {
820 v.push(*i);
821 Ok(())
822 }
823 _ => Err(AvroError::InvalidArgument(
824 "Default for int32/date32/time-millis must be int".to_string(),
825 )),
826 },
827 #[cfg(feature = "avro_custom_types")]
828 Self::DurationSecond(v)
829 | Self::DurationMillisecond(v)
830 | Self::DurationMicrosecond(v)
831 | Self::DurationNanosecond(v) => match lit {
832 AvroLiteral::Long(i) => {
833 v.push(*i);
834 Ok(())
835 }
836 _ => Err(AvroError::InvalidArgument(
837 "Default for duration long must be long".to_string(),
838 )),
839 },
840 #[cfg(feature = "avro_custom_types")]
841 Self::Int8(v) => match lit {
842 AvroLiteral::Int(i) => {
843 let x = i8::try_from(*i).map_err(|_| {
844 AvroError::InvalidArgument(format!(
845 "Default for int8 out of range for i8: {i}"
846 ))
847 })?;
848 v.push(x);
849 Ok(())
850 }
851 _ => Err(AvroError::InvalidArgument(
852 "Default for int8 must be int".to_string(),
853 )),
854 },
855 #[cfg(feature = "avro_custom_types")]
856 Self::Int16(v) => match lit {
857 AvroLiteral::Int(i) => {
858 let x = i16::try_from(*i).map_err(|_| {
859 AvroError::InvalidArgument(format!(
860 "Default for int16 out of range for i16: {i}"
861 ))
862 })?;
863 v.push(x);
864 Ok(())
865 }
866 _ => Err(AvroError::InvalidArgument(
867 "Default for int16 must be int".to_string(),
868 )),
869 },
870 #[cfg(feature = "avro_custom_types")]
871 Self::UInt8(v) => match lit {
872 AvroLiteral::Int(i) => {
873 let x = u8::try_from(*i).map_err(|_| {
874 AvroError::InvalidArgument(format!(
875 "Default for uint8 out of range for u8: {i}"
876 ))
877 })?;
878 v.push(x);
879 Ok(())
880 }
881 _ => Err(AvroError::InvalidArgument(
882 "Default for uint8 must be int".to_string(),
883 )),
884 },
885 #[cfg(feature = "avro_custom_types")]
886 Self::UInt16(v) => match lit {
887 AvroLiteral::Int(i) => {
888 let x = u16::try_from(*i).map_err(|_| {
889 AvroError::InvalidArgument(format!(
890 "Default for uint16 out of range for u16: {i}"
891 ))
892 })?;
893 v.push(x);
894 Ok(())
895 }
896 _ => Err(AvroError::InvalidArgument(
897 "Default for uint16 must be int".to_string(),
898 )),
899 },
900 #[cfg(feature = "avro_custom_types")]
901 Self::UInt32(v) => match lit {
902 AvroLiteral::Long(i) => {
903 let x = u32::try_from(*i).map_err(|_| {
904 AvroError::InvalidArgument(format!(
905 "Default for uint32 out of range for u32: {i}"
906 ))
907 })?;
908 v.push(x);
909 Ok(())
910 }
911 _ => Err(AvroError::InvalidArgument(
912 "Default for uint32 must be long".to_string(),
913 )),
914 },
915 #[cfg(feature = "avro_custom_types")]
916 Self::UInt64(v) => match lit {
917 AvroLiteral::Bytes(b) => {
918 if b.len() != 8 {
919 return Err(AvroError::InvalidArgument(format!(
920 "uint64 default must be exactly 8 bytes, got {}",
921 b.len()
922 )));
923 }
924 v.push(u64::from_le_bytes([
925 b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7],
926 ]));
927 Ok(())
928 }
929 _ => Err(AvroError::InvalidArgument(
930 "Default for uint64 must be bytes (8-byte LE)".to_string(),
931 )),
932 },
933 #[cfg(feature = "avro_custom_types")]
934 Self::Float16(v) => match lit {
935 AvroLiteral::Bytes(b) => {
936 if b.len() != 2 {
937 return Err(AvroError::InvalidArgument(format!(
938 "float16 default must be exactly 2 bytes, got {}",
939 b.len()
940 )));
941 }
942 v.push(u16::from_le_bytes([b[0], b[1]]));
943 Ok(())
944 }
945 _ => Err(AvroError::InvalidArgument(
946 "Default for float16 must be bytes (2-byte LE IEEE-754)".to_string(),
947 )),
948 },
949 #[cfg(feature = "avro_custom_types")]
950 Self::Date64(v) | Self::TimeNanos(v) | Self::TimestampSecs(_, v) => match lit {
951 AvroLiteral::Long(i) => {
952 v.push(*i);
953 Ok(())
954 }
955 _ => Err(AvroError::InvalidArgument(
956 "Default for date64/time-nanos/timestamp-secs must be long".to_string(),
957 )),
958 },
959 #[cfg(feature = "avro_custom_types")]
960 Self::Time32Secs(v) => match lit {
961 AvroLiteral::Int(i) => {
962 v.push(*i);
963 Ok(())
964 }
965 _ => Err(AvroError::InvalidArgument(
966 "Default for time32-secs must be int".to_string(),
967 )),
968 },
969 #[cfg(feature = "avro_custom_types")]
970 Self::IntervalYearMonth(v) => match lit {
971 AvroLiteral::Bytes(b) => {
972 if b.len() != 4 {
973 return Err(AvroError::InvalidArgument(format!(
974 "interval-year-month default must be exactly 4 bytes, got {}",
975 b.len()
976 )));
977 }
978 v.push(i32::from_le_bytes([b[0], b[1], b[2], b[3]]));
979 Ok(())
980 }
981 _ => Err(AvroError::InvalidArgument(
982 "Default for interval-year-month must be bytes (4-byte LE)".to_string(),
983 )),
984 },
985 #[cfg(feature = "avro_custom_types")]
986 Self::IntervalMonthDayNano(v) => match lit {
987 AvroLiteral::Bytes(b) => {
988 if b.len() != 16 {
989 return Err(AvroError::InvalidArgument(format!(
990 "interval-month-day-nano default must be exactly 16 bytes, got {}",
991 b.len()
992 )));
993 }
994 let months = i32::from_le_bytes([b[0], b[1], b[2], b[3]]);
995 let days = i32::from_le_bytes([b[4], b[5], b[6], b[7]]);
996 let nanos =
997 i64::from_le_bytes([b[8], b[9], b[10], b[11], b[12], b[13], b[14], b[15]]);
998 v.push(IntervalMonthDayNano::new(months, days, nanos));
999 Ok(())
1000 }
1001 _ => Err(AvroError::InvalidArgument(
1002 "Default for interval-month-day-nano must be bytes (16-byte LE)".to_string(),
1003 )),
1004 },
1005 #[cfg(feature = "avro_custom_types")]
1006 Self::IntervalDayTime(v) => match lit {
1007 AvroLiteral::Bytes(b) => {
1008 if b.len() != 8 {
1009 return Err(AvroError::InvalidArgument(format!(
1010 "interval-day-time default must be exactly 8 bytes, got {}",
1011 b.len()
1012 )));
1013 }
1014 let days = i32::from_le_bytes([b[0], b[1], b[2], b[3]]);
1015 let milliseconds = i32::from_le_bytes([b[4], b[5], b[6], b[7]]);
1016 v.push(IntervalDayTime::new(days, milliseconds));
1017 Ok(())
1018 }
1019 _ => Err(AvroError::InvalidArgument(
1020 "Default for interval-day-time must be bytes (8-byte LE)".to_string(),
1021 )),
1022 },
1023 Self::Int64(v)
1024 | Self::Int32ToInt64(v)
1025 | Self::TimeMicros(v)
1026 | Self::TimestampMillis(_, v)
1027 | Self::TimestampMicros(_, v)
1028 | Self::TimestampNanos(_, v) => match lit {
1029 AvroLiteral::Long(i) => {
1030 v.push(*i);
1031 Ok(())
1032 }
1033 AvroLiteral::Int(i) => {
1034 v.push(*i as i64);
1035 Ok(())
1036 }
1037 _ => Err(AvroError::InvalidArgument(
1038 "Default for long/time-micros/timestamp must be long or int".to_string(),
1039 )),
1040 },
1041 Self::Float32(v) | Self::Int32ToFloat32(v) | Self::Int64ToFloat32(v) => match lit {
1042 AvroLiteral::Float(f) => {
1043 v.push(*f);
1044 Ok(())
1045 }
1046 _ => Err(AvroError::InvalidArgument(
1047 "Default for float must be float".to_string(),
1048 )),
1049 },
1050 Self::Float64(v)
1051 | Self::Int32ToFloat64(v)
1052 | Self::Int64ToFloat64(v)
1053 | Self::Float32ToFloat64(v) => match lit {
1054 AvroLiteral::Double(f) => {
1055 v.push(*f);
1056 Ok(())
1057 }
1058 _ => Err(AvroError::InvalidArgument(
1059 "Default for double must be double".to_string(),
1060 )),
1061 },
1062 Self::Binary(offsets, values) | Self::StringToBytes(offsets, values) => match lit {
1063 AvroLiteral::Bytes(b) => {
1064 offsets.push_length(b.len());
1065 values.extend_from_slice(b);
1066 Ok(())
1067 }
1068 _ => Err(AvroError::InvalidArgument(
1069 "Default for bytes must be bytes".to_string(),
1070 )),
1071 },
1072 Self::BytesToString(offsets, values)
1073 | Self::String(offsets, values)
1074 | Self::StringView(offsets, values) => match lit {
1075 AvroLiteral::String(s) => {
1076 let b = s.as_bytes();
1077 offsets.push_length(b.len());
1078 values.extend_from_slice(b);
1079 Ok(())
1080 }
1081 _ => Err(AvroError::InvalidArgument(
1082 "Default for string must be string".to_string(),
1083 )),
1084 },
1085 Self::Uuid(values) => match lit {
1086 AvroLiteral::String(s) => {
1087 let uuid = Uuid::try_parse(s).map_err(|e| {
1088 AvroError::InvalidArgument(format!("Invalid UUID default: {s} ({e})"))
1089 })?;
1090 values.extend_from_slice(uuid.as_bytes());
1091 Ok(())
1092 }
1093 _ => Err(AvroError::InvalidArgument(
1094 "Default for uuid must be string".to_string(),
1095 )),
1096 },
1097 Self::Fixed(sz, accum) => match lit {
1098 AvroLiteral::Bytes(b) => {
1099 if b.len() != *sz as usize {
1100 return Err(AvroError::InvalidArgument(format!(
1101 "Fixed default length {} does not match size {sz}",
1102 b.len(),
1103 )));
1104 }
1105 accum.extend_from_slice(b);
1106 Ok(())
1107 }
1108 _ => Err(AvroError::InvalidArgument(
1109 "Default for fixed must be bytes".to_string(),
1110 )),
1111 },
1112 #[cfg(feature = "small_decimals")]
1113 Self::Decimal32(_, _, _, builder) => {
1114 append_decimal_default!(lit, builder, 4, i32, "decimal32")
1115 }
1116 #[cfg(feature = "small_decimals")]
1117 Self::Decimal64(_, _, _, builder) => {
1118 append_decimal_default!(lit, builder, 8, i64, "decimal64")
1119 }
1120 Self::Decimal128(_, _, _, builder) => {
1121 append_decimal_default!(lit, builder, 16, i128, "decimal128")
1122 }
1123 Self::Decimal256(_, _, _, builder) => {
1124 append_decimal_default!(lit, builder, 32, i256, "decimal256")
1125 }
1126 Self::Duration(builder) => match lit {
1127 AvroLiteral::Bytes(b) => {
1128 if b.len() != 12 {
1129 return Err(AvroError::InvalidArgument(format!(
1130 "Duration default must be exactly 12 bytes, got {}",
1131 b.len()
1132 )));
1133 }
1134 let months = u32::from_le_bytes([b[0], b[1], b[2], b[3]]);
1135 let days = u32::from_le_bytes([b[4], b[5], b[6], b[7]]);
1136 let millis = u32::from_le_bytes([b[8], b[9], b[10], b[11]]);
1137 let nanos = (millis as i64) * 1_000_000;
1138 builder.append_value(IntervalMonthDayNano::new(
1139 months as i32,
1140 days as i32,
1141 nanos,
1142 ));
1143 Ok(())
1144 }
1145 _ => Err(AvroError::InvalidArgument(
1146 "Default for duration must be 12-byte little-endian months/days/millis"
1147 .to_string(),
1148 )),
1149 },
1150 Self::Array(_, offsets, inner) => match lit {
1151 AvroLiteral::Array(items) => {
1152 offsets.push_length(items.len());
1153 for item in items {
1154 inner.append_default(item)?;
1155 }
1156 Ok(())
1157 }
1158 _ => Err(AvroError::InvalidArgument(
1159 "Default for array must be an array literal".to_string(),
1160 )),
1161 },
1162 Self::Map(_, koff, moff, kdata, valdec) => match lit {
1163 AvroLiteral::Map(entries) => {
1164 moff.push_length(entries.len());
1165 for (k, v) in entries {
1166 let kb = k.as_bytes();
1167 koff.push_length(kb.len());
1168 kdata.extend_from_slice(kb);
1169 valdec.append_default(v)?;
1170 }
1171 Ok(())
1172 }
1173 _ => Err(AvroError::InvalidArgument(
1174 "Default for map must be a map/object literal".to_string(),
1175 )),
1176 },
1177 Self::Enum(indices, symbols, _) => match lit {
1178 AvroLiteral::Enum(sym) => {
1179 let pos = symbols.iter().position(|s| s == sym).ok_or_else(|| {
1180 AvroError::InvalidArgument(format!(
1181 "Enum default symbol {sym:?} not in reader symbols"
1182 ))
1183 })?;
1184 indices.push(pos as i32);
1185 Ok(())
1186 }
1187 _ => Err(AvroError::InvalidArgument(
1188 "Default for enum must be a symbol".to_string(),
1189 )),
1190 },
1191 #[cfg(feature = "avro_custom_types")]
1192 Self::RunEndEncoded(_, len, inner) => {
1193 *len += 1;
1194 inner.append_default(lit)
1195 }
1196 Self::Union(u) => u.append_default(lit),
1197 Self::Record {
1198 fields: field_meta,
1199 decoders,
1200 defaults: field_defaults,
1201 row_count,
1202 ..
1203 } => match lit {
1204 AvroLiteral::Map(entries) => {
1205 for (i, dec) in decoders.iter_mut().enumerate() {
1206 let name = field_meta[i].name();
1207 if let Some(sub) = entries.get(name) {
1208 dec.append_default(sub)?;
1209 } else if let Some(default_literal) = field_defaults[i].as_ref() {
1210 dec.append_default(default_literal)?;
1211 } else {
1212 dec.append_null()?;
1213 }
1214 }
1215 *row_count += 1;
1216 Ok(())
1217 }
1218 AvroLiteral::Null => {
1219 for (i, dec) in decoders.iter_mut().enumerate() {
1220 if let Some(default_literal) = field_defaults[i].as_ref() {
1221 dec.append_default(default_literal)?;
1222 } else {
1223 dec.append_null()?;
1224 }
1225 }
1226 *row_count += 1;
1227 Ok(())
1228 }
1229 _ => Err(AvroError::InvalidArgument(
1230 "Default for record must be a map/object or null".to_string(),
1231 )),
1232 },
1233 }
1234 }
1235
1236 fn decode(&mut self, buf: &mut AvroCursor<'_>) -> Result<(), AvroError> {
1238 match self {
1239 Self::Null(x) => *x += 1,
1240 Self::Boolean(values) => values.append(buf.get_bool()?),
1241 Self::Int32(values) | Self::Date32(values) | Self::TimeMillis(values) => {
1242 values.push(buf.get_int()?)
1243 }
1244 Self::Int64(values)
1245 | Self::TimeMicros(values)
1246 | Self::TimestampMillis(_, values)
1247 | Self::TimestampMicros(_, values)
1248 | Self::TimestampNanos(_, values) => values.push(buf.get_long()?),
1249 #[cfg(feature = "avro_custom_types")]
1250 Self::DurationSecond(values)
1251 | Self::DurationMillisecond(values)
1252 | Self::DurationMicrosecond(values)
1253 | Self::DurationNanosecond(values) => values.push(buf.get_long()?),
1254 #[cfg(feature = "avro_custom_types")]
1255 Self::Int8(values) => {
1256 let raw = buf.get_int()?;
1257 let x = i8::try_from(raw).map_err(|_| {
1258 AvroError::ParseError(format!("int8 value {raw} out of range for i8"))
1259 })?;
1260 values.push(x);
1261 }
1262 #[cfg(feature = "avro_custom_types")]
1263 Self::Int16(values) => {
1264 let raw = buf.get_int()?;
1265 let x = i16::try_from(raw).map_err(|_| {
1266 AvroError::ParseError(format!("int16 value {raw} out of range for i16"))
1267 })?;
1268 values.push(x);
1269 }
1270 #[cfg(feature = "avro_custom_types")]
1271 Self::UInt8(values) => {
1272 let raw = buf.get_int()?;
1273 let x = u8::try_from(raw).map_err(|_| {
1274 AvroError::ParseError(format!("uint8 value {raw} out of range for u8"))
1275 })?;
1276 values.push(x);
1277 }
1278 #[cfg(feature = "avro_custom_types")]
1279 Self::UInt16(values) => {
1280 let raw = buf.get_int()?;
1281 let x = u16::try_from(raw).map_err(|_| {
1282 AvroError::ParseError(format!("uint16 value {raw} out of range for u16"))
1283 })?;
1284 values.push(x);
1285 }
1286 #[cfg(feature = "avro_custom_types")]
1287 Self::UInt32(values) => {
1288 let raw = buf.get_long()?;
1289 let x = u32::try_from(raw).map_err(|_| {
1290 AvroError::ParseError(format!("uint32 value {raw} out of range for u32"))
1291 })?;
1292 values.push(x);
1293 }
1294 #[cfg(feature = "avro_custom_types")]
1295 Self::UInt64(values) => {
1296 let b = buf.get_fixed(8)?;
1297 values.push(u64::from_le_bytes([
1298 b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7],
1299 ]));
1300 }
1301 #[cfg(feature = "avro_custom_types")]
1302 Self::Float16(values) => {
1303 let b = buf.get_fixed(2)?;
1304 values.push(u16::from_le_bytes([b[0], b[1]]));
1305 }
1306 #[cfg(feature = "avro_custom_types")]
1307 Self::Date64(values) | Self::TimeNanos(values) | Self::TimestampSecs(_, values) => {
1308 values.push(buf.get_long()?)
1309 }
1310 #[cfg(feature = "avro_custom_types")]
1311 Self::Time32Secs(values) => values.push(buf.get_int()?),
1312 #[cfg(feature = "avro_custom_types")]
1313 Self::IntervalYearMonth(values) => {
1314 let b = buf.get_fixed(4)?;
1315 values.push(i32::from_le_bytes([b[0], b[1], b[2], b[3]]));
1316 }
1317 #[cfg(feature = "avro_custom_types")]
1318 Self::IntervalMonthDayNano(values) => {
1319 let b = buf.get_fixed(16)?;
1320 let months = i32::from_le_bytes([b[0], b[1], b[2], b[3]]);
1321 let days = i32::from_le_bytes([b[4], b[5], b[6], b[7]]);
1322 let nanos =
1323 i64::from_le_bytes([b[8], b[9], b[10], b[11], b[12], b[13], b[14], b[15]]);
1324 values.push(IntervalMonthDayNano::new(months, days, nanos));
1325 }
1326 #[cfg(feature = "avro_custom_types")]
1327 Self::IntervalDayTime(values) => {
1328 let b = buf.get_fixed(8)?;
1329 let days = i32::from_le_bytes([b[0], b[1], b[2], b[3]]);
1331 let milliseconds = i32::from_le_bytes([b[4], b[5], b[6], b[7]]);
1332 values.push(IntervalDayTime::new(days, milliseconds));
1333 }
1334 Self::Float32(values) => values.push(buf.get_float()?),
1335 Self::Float64(values) => values.push(buf.get_double()?),
1336 Self::Int32ToInt64(values) => values.push(buf.get_int()? as i64),
1337 Self::Int32ToFloat32(values) => values.push(buf.get_int()? as f32),
1338 Self::Int32ToFloat64(values) => values.push(buf.get_int()? as f64),
1339 Self::Int64ToFloat32(values) => values.push(buf.get_long()? as f32),
1340 Self::Int64ToFloat64(values) => values.push(buf.get_long()? as f64),
1341 Self::Float32ToFloat64(values) => values.push(buf.get_float()? as f64),
1342 Self::StringToBytes(offsets, values)
1343 | Self::BytesToString(offsets, values)
1344 | Self::Binary(offsets, values)
1345 | Self::String(offsets, values)
1346 | Self::StringView(offsets, values) => {
1347 let data = buf.get_bytes()?;
1348 offsets.push_length(data.len());
1349 values.extend_from_slice(data);
1350 }
1351 Self::Uuid(values) => {
1352 let s_bytes = buf.get_bytes()?;
1353 let s = std::str::from_utf8(s_bytes).map_err(|e| {
1354 AvroError::ParseError(format!("UUID bytes are not valid UTF-8: {e}"))
1355 })?;
1356 let uuid = Uuid::try_parse(s)
1357 .map_err(|e| AvroError::ParseError(format!("Failed to parse uuid: {e}")))?;
1358 values.extend_from_slice(uuid.as_bytes());
1359 }
1360 Self::Array(_, off, encoding) => {
1361 let total_items = read_blocks(buf, |cursor| encoding.decode(cursor))?;
1362 off.push_length(total_items);
1363 }
1364 Self::Record {
1365 decoders,
1366 projector: None,
1367 row_count,
1368 ..
1369 } => {
1370 for encoding in decoders {
1371 encoding.decode(buf)?;
1372 }
1373 *row_count += 1;
1374 }
1375 Self::Record {
1376 decoders,
1377 projector: Some(proj),
1378 row_count,
1379 ..
1380 } => {
1381 proj.project_record(buf, decoders)?;
1382 *row_count += 1;
1383 }
1384 Self::Map(_, koff, moff, kdata, valdec) => {
1385 let newly_added = read_blocks(buf, |cur| {
1386 let kb = cur.get_bytes()?;
1387 koff.push_length(kb.len());
1388 kdata.extend_from_slice(kb);
1389 valdec.decode(cur)
1390 })?;
1391 moff.push_length(newly_added);
1392 }
1393 Self::Fixed(sz, accum) => {
1394 let fx = buf.get_fixed(*sz as usize)?;
1395 accum.extend_from_slice(fx);
1396 }
1397 #[cfg(feature = "small_decimals")]
1398 Self::Decimal32(_, _, size, builder) => {
1399 decode_decimal!(size, buf, builder, 4, i32);
1400 }
1401 #[cfg(feature = "small_decimals")]
1402 Self::Decimal64(_, _, size, builder) => {
1403 decode_decimal!(size, buf, builder, 8, i64);
1404 }
1405 Self::Decimal128(_, _, size, builder) => {
1406 decode_decimal!(size, buf, builder, 16, i128);
1407 }
1408 Self::Decimal256(_, _, size, builder) => {
1409 decode_decimal!(size, buf, builder, 32, i256);
1410 }
1411 Self::Enum(indices, _, None) => {
1412 indices.push(buf.get_int()?);
1413 }
1414 Self::Enum(indices, _, Some(res)) => {
1415 let raw = buf.get_int()?;
1416 let resolved = res.resolve(raw)?;
1417 indices.push(resolved);
1418 }
1419 Self::Duration(builder) => {
1420 let b = buf.get_fixed(12)?;
1421 let months = u32::from_le_bytes([b[0], b[1], b[2], b[3]]);
1422 let days = u32::from_le_bytes([b[4], b[5], b[6], b[7]]);
1423 let millis = u32::from_le_bytes([b[8], b[9], b[10], b[11]]);
1424 let nanos = (millis as i64) * 1_000_000;
1425 builder.append_value(IntervalMonthDayNano::new(months as i32, days as i32, nanos));
1426 }
1427 #[cfg(feature = "avro_custom_types")]
1428 Self::RunEndEncoded(_, len, inner) => {
1429 *len += 1;
1430 inner.decode(buf)?;
1431 }
1432 Self::Union(u) => u.decode(buf)?,
1433 Self::Nullable(nullable) => {
1434 let is_not_null = match &nullable.plan {
1435 NullablePlan::FromSingle { .. } => true,
1436 NullablePlan::ReadTag { nullability, .. } => {
1437 let branch = buf.read_vlq()?;
1438 match *nullability {
1439 Nullability::NullFirst => branch != 0,
1440 Nullability::NullSecond => branch == 0,
1441 }
1442 }
1443 };
1444
1445 if is_not_null {
1446 nullable.materialize_pending()?;
1447 let resolution = match &nullable.plan {
1448 NullablePlan::FromSingle { resolution }
1449 | NullablePlan::ReadTag { resolution, .. } => resolution,
1450 };
1451 nullable.values.decode_with_resolution(buf, resolution)?;
1453 nullable.validity.append_non_null();
1454 } else {
1455 nullable.pending_nulls += 1;
1456 }
1457 }
1458 }
1459 Ok(())
1460 }
1461
1462 fn decode_with_promotion(
1463 &mut self,
1464 buf: &mut AvroCursor<'_>,
1465 promotion: Promotion,
1466 ) -> Result<(), AvroError> {
1467 #[cfg(feature = "avro_custom_types")]
1468 if let Self::RunEndEncoded(_, len, inner) = self {
1469 *len += 1;
1470 return inner.decode_with_promotion(buf, promotion);
1471 }
1472
1473 macro_rules! promote_numeric_to {
1474 ($variant:ident, $getter:ident, $to:ty) => {{
1475 match self {
1476 Self::$variant(v) => {
1477 let x = buf.$getter()?;
1478 v.push(x as $to);
1479 Ok(())
1480 }
1481 other => Err(AvroError::ParseError(format!(
1482 "Promotion {promotion} target mismatch: expected {}, got {}",
1483 stringify!($variant),
1484 <Self as ::std::convert::AsRef<str>>::as_ref(other)
1485 ))),
1486 }
1487 }};
1488 }
1489 match promotion {
1490 Promotion::Direct => self.decode(buf),
1491 Promotion::IntToLong => promote_numeric_to!(Int64, get_int, i64),
1492 Promotion::IntToFloat => promote_numeric_to!(Float32, get_int, f32),
1493 Promotion::IntToDouble => promote_numeric_to!(Float64, get_int, f64),
1494 Promotion::LongToFloat => promote_numeric_to!(Float32, get_long, f32),
1495 Promotion::LongToDouble => promote_numeric_to!(Float64, get_long, f64),
1496 Promotion::FloatToDouble => promote_numeric_to!(Float64, get_float, f64),
1497 Promotion::StringToBytes => match self {
1498 Self::Binary(offsets, values) | Self::StringToBytes(offsets, values) => {
1499 let data = buf.get_bytes()?;
1500 offsets.push_length(data.len());
1501 values.extend_from_slice(data);
1502 Ok(())
1503 }
1504 other => Err(AvroError::ParseError(format!(
1505 "Promotion {promotion} target mismatch: expected bytes (Binary/StringToBytes), got {}",
1506 <Self as AsRef<str>>::as_ref(other)
1507 ))),
1508 },
1509 Promotion::BytesToString => match self {
1510 Self::String(offsets, values)
1511 | Self::StringView(offsets, values)
1512 | Self::BytesToString(offsets, values) => {
1513 let data = buf.get_bytes()?;
1514 offsets.push_length(data.len());
1515 values.extend_from_slice(data);
1516 Ok(())
1517 }
1518 other => Err(AvroError::ParseError(format!(
1519 "Promotion {promotion} target mismatch: expected string (String/StringView/BytesToString), got {}",
1520 <Self as AsRef<str>>::as_ref(other)
1521 ))),
1522 },
1523 }
1524 }
1525
1526 fn decode_with_resolution<'d>(
1527 &'d mut self,
1528 buf: &mut AvroCursor<'_>,
1529 resolution: &'d ResolutionPlan,
1530 ) -> Result<(), AvroError> {
1531 #[cfg(feature = "avro_custom_types")]
1532 if let Self::RunEndEncoded(_, len, inner) = self {
1533 *len += 1;
1534 return inner.decode_with_resolution(buf, resolution);
1535 }
1536
1537 match resolution {
1538 ResolutionPlan::Promotion(promotion) => {
1539 let promotion = *promotion;
1540 self.decode_with_promotion(buf, promotion)
1541 }
1542 ResolutionPlan::DefaultValue(lit) => self.append_default(lit),
1543 ResolutionPlan::EnumMapping(res) => {
1544 let Self::Enum(indices, _, _) = self else {
1545 return Err(AvroError::SchemaError(
1546 "enum mapping resolution provided for non-enum decoder".into(),
1547 ));
1548 };
1549 let raw = buf.get_int()?;
1550 let resolved = res.resolve(raw)?;
1551 indices.push(resolved);
1552 Ok(())
1553 }
1554 ResolutionPlan::Record(proj) => {
1555 let Self::Record {
1556 decoders,
1557 row_count,
1558 ..
1559 } = self
1560 else {
1561 return Err(AvroError::SchemaError(
1562 "record projection provided for non-record decoder".into(),
1563 ));
1564 };
1565 proj.project_record(buf, decoders)?;
1566 *row_count += 1;
1567 Ok(())
1568 }
1569 }
1570 }
1571
1572 fn flush(&mut self, nulls: Option<NullBuffer>) -> Result<ArrayRef, AvroError> {
1574 Ok(match self {
1575 Self::Nullable(nullable) => {
1576 nullable.materialize_pending()?;
1577 nullable.values.flush(nullable.validity.finish())?
1578 }
1579 Self::Null(size) => Arc::new(NullArray::new(std::mem::replace(size, 0))),
1580 Self::Boolean(b) => Arc::new(BooleanArray::new(b.finish(), nulls)),
1581 Self::Int32(values) => Arc::new(flush_primitive::<Int32Type>(values, nulls)),
1582 Self::Date32(values) => Arc::new(flush_primitive::<Date32Type>(values, nulls)),
1583 Self::Int64(values) => Arc::new(flush_primitive::<Int64Type>(values, nulls)),
1584 Self::TimeMillis(values) => {
1585 Arc::new(flush_primitive::<Time32MillisecondType>(values, nulls))
1586 }
1587 Self::TimeMicros(values) => {
1588 Arc::new(flush_primitive::<Time64MicrosecondType>(values, nulls))
1589 }
1590 Self::TimestampMillis(tz, values) => Arc::new(
1591 flush_primitive::<TimestampMillisecondType>(values, nulls)
1592 .with_timezone_opt(tz.as_ref().map(|tz| tz.to_string())),
1593 ),
1594 Self::TimestampMicros(tz, values) => Arc::new(
1595 flush_primitive::<TimestampMicrosecondType>(values, nulls)
1596 .with_timezone_opt(tz.as_ref().map(|tz| tz.to_string())),
1597 ),
1598 Self::TimestampNanos(tz, values) => Arc::new(
1599 flush_primitive::<TimestampNanosecondType>(values, nulls)
1600 .with_timezone_opt(tz.as_ref().map(|tz| tz.to_string())),
1601 ),
1602 #[cfg(feature = "avro_custom_types")]
1603 Self::DurationSecond(values) => {
1604 Arc::new(flush_primitive::<DurationSecondType>(values, nulls))
1605 }
1606 #[cfg(feature = "avro_custom_types")]
1607 Self::DurationMillisecond(values) => {
1608 Arc::new(flush_primitive::<DurationMillisecondType>(values, nulls))
1609 }
1610 #[cfg(feature = "avro_custom_types")]
1611 Self::DurationMicrosecond(values) => {
1612 Arc::new(flush_primitive::<DurationMicrosecondType>(values, nulls))
1613 }
1614 #[cfg(feature = "avro_custom_types")]
1615 Self::DurationNanosecond(values) => {
1616 Arc::new(flush_primitive::<DurationNanosecondType>(values, nulls))
1617 }
1618 #[cfg(feature = "avro_custom_types")]
1619 Self::Int8(values) => Arc::new(flush_primitive::<Int8Type>(values, nulls)),
1620 #[cfg(feature = "avro_custom_types")]
1621 Self::Int16(values) => Arc::new(flush_primitive::<Int16Type>(values, nulls)),
1622 #[cfg(feature = "avro_custom_types")]
1623 Self::UInt8(values) => Arc::new(flush_primitive::<UInt8Type>(values, nulls)),
1624 #[cfg(feature = "avro_custom_types")]
1625 Self::UInt16(values) => Arc::new(flush_primitive::<UInt16Type>(values, nulls)),
1626 #[cfg(feature = "avro_custom_types")]
1627 Self::UInt32(values) => Arc::new(flush_primitive::<UInt32Type>(values, nulls)),
1628 #[cfg(feature = "avro_custom_types")]
1629 Self::UInt64(values) => Arc::new(flush_primitive::<UInt64Type>(values, nulls)),
1630 #[cfg(feature = "avro_custom_types")]
1631 Self::Float16(values) => {
1632 let len = values.len();
1635 let buf: Buffer = std::mem::take(values).into();
1636 let scalar_buf = ScalarBuffer::new(buf, 0, len);
1637 Arc::new(Float16Array::new(scalar_buf, nulls))
1638 }
1639 #[cfg(feature = "avro_custom_types")]
1640 Self::Date64(values) => Arc::new(flush_primitive::<Date64Type>(values, nulls)),
1641 #[cfg(feature = "avro_custom_types")]
1642 Self::TimeNanos(values) => {
1643 Arc::new(flush_primitive::<Time64NanosecondType>(values, nulls))
1644 }
1645 #[cfg(feature = "avro_custom_types")]
1646 Self::Time32Secs(values) => {
1647 Arc::new(flush_primitive::<Time32SecondType>(values, nulls))
1648 }
1649 #[cfg(feature = "avro_custom_types")]
1650 Self::TimestampSecs(is_utc, values) => Arc::new(
1651 flush_primitive::<TimestampSecondType>(values, nulls)
1652 .with_timezone_opt(is_utc.then(|| "+00:00")),
1653 ),
1654 #[cfg(feature = "avro_custom_types")]
1655 Self::IntervalYearMonth(values) => {
1656 Arc::new(flush_primitive::<IntervalYearMonthType>(values, nulls))
1657 }
1658 #[cfg(feature = "avro_custom_types")]
1659 Self::IntervalMonthDayNano(values) => {
1660 Arc::new(flush_primitive::<IntervalMonthDayNanoType>(values, nulls))
1661 }
1662 #[cfg(feature = "avro_custom_types")]
1663 Self::IntervalDayTime(values) => {
1664 Arc::new(flush_primitive::<IntervalDayTimeType>(values, nulls))
1665 }
1666 Self::Float32(values) => Arc::new(flush_primitive::<Float32Type>(values, nulls)),
1667 Self::Float64(values) => Arc::new(flush_primitive::<Float64Type>(values, nulls)),
1668 Self::Int32ToInt64(values) => Arc::new(flush_primitive::<Int64Type>(values, nulls)),
1669 Self::Int32ToFloat32(values) | Self::Int64ToFloat32(values) => {
1670 Arc::new(flush_primitive::<Float32Type>(values, nulls))
1671 }
1672 Self::Int32ToFloat64(values)
1673 | Self::Int64ToFloat64(values)
1674 | Self::Float32ToFloat64(values) => {
1675 Arc::new(flush_primitive::<Float64Type>(values, nulls))
1676 }
1677 Self::StringToBytes(offsets, values) | Self::Binary(offsets, values) => {
1678 let offsets = flush_offsets(offsets);
1679 let values = flush_values(values).into();
1680 Arc::new(BinaryArray::try_new(offsets, values, nulls)?)
1681 }
1682 Self::BytesToString(offsets, values) | Self::String(offsets, values) => {
1683 let offsets = flush_offsets(offsets);
1684 let values = flush_values(values).into();
1685 Arc::new(StringArray::try_new(offsets, values, nulls)?)
1686 }
1687 Self::StringView(offsets, values) => {
1688 let offsets = flush_offsets(offsets);
1689 let values = flush_values(values);
1690 let array = StringArray::try_new(offsets, values.into(), nulls)?;
1691 Arc::new(StringViewArray::from(&array))
1692 }
1693 Self::Array(field, offsets, values) => {
1694 let values = values.flush(None)?;
1695 let offsets = flush_offsets(offsets);
1696 Arc::new(ListArray::try_new(field.clone(), offsets, values, nulls)?)
1697 }
1698 Self::Record {
1699 fields,
1700 decoders,
1701 row_count,
1702 ..
1703 } => {
1704 let arrays = decoders
1705 .iter_mut()
1706 .map(|x| x.flush(None))
1707 .collect::<Result<Vec<_>, _>>()?;
1708 Arc::new(StructArray::try_new_with_length(
1711 fields.clone(),
1712 arrays,
1713 nulls,
1714 std::mem::replace(row_count, 0),
1715 )?)
1716 }
1717 Self::Map(map_field, k_off, m_off, kdata, valdec) => {
1718 let moff = flush_offsets(m_off);
1719 let koff = flush_offsets(k_off);
1720 let kd = flush_values(kdata).into();
1721 let val_arr = valdec.flush(None)?;
1722 let key_arr = StringArray::try_new(koff, kd, None)?;
1723 if key_arr.len() != val_arr.len() {
1724 return Err(AvroError::InvalidArgument(format!(
1725 "Map keys length ({}) != map values length ({})",
1726 key_arr.len(),
1727 val_arr.len()
1728 )));
1729 }
1730 let final_len = moff.len() - 1;
1731 if let Some(n) = &nulls
1732 && n.len() != final_len
1733 {
1734 return Err(AvroError::InvalidArgument(format!(
1735 "Map array null buffer length {} != final map length {final_len}",
1736 n.len()
1737 )));
1738 }
1739 let entries_fields = match map_field.data_type() {
1740 DataType::Struct(fields) => fields.clone(),
1741 other => {
1742 return Err(AvroError::InvalidArgument(format!(
1743 "Map entries field must be a Struct, got {other:?}"
1744 )));
1745 }
1746 };
1747 let entries_struct =
1748 StructArray::try_new(entries_fields, vec![Arc::new(key_arr), val_arr], None)?;
1749 let map_arr =
1750 MapArray::try_new(map_field.clone(), moff, entries_struct, nulls, false)?;
1751 Arc::new(map_arr)
1752 }
1753 Self::Fixed(sz, accum) => {
1754 let b: Buffer = flush_values(accum).into();
1755 let arr = FixedSizeBinaryArray::try_new(*sz, b, nulls)
1756 .map_err(|e| AvroError::ParseError(e.to_string()))?;
1757 Arc::new(arr)
1758 }
1759 Self::Uuid(values) => {
1760 let arr = FixedSizeBinaryArray::try_new(16, std::mem::take(values).into(), nulls)
1761 .map_err(|e| AvroError::ParseError(e.to_string()))?;
1762 Arc::new(arr)
1763 }
1764 #[cfg(feature = "small_decimals")]
1765 Self::Decimal32(precision, scale, _, builder) => {
1766 flush_decimal!(builder, precision, scale, nulls, Decimal32Array)
1767 }
1768 #[cfg(feature = "small_decimals")]
1769 Self::Decimal64(precision, scale, _, builder) => {
1770 flush_decimal!(builder, precision, scale, nulls, Decimal64Array)
1771 }
1772 Self::Decimal128(precision, scale, _, builder) => {
1773 flush_decimal!(builder, precision, scale, nulls, Decimal128Array)
1774 }
1775 Self::Decimal256(precision, scale, _, builder) => {
1776 flush_decimal!(builder, precision, scale, nulls, Decimal256Array)
1777 }
1778 Self::Enum(indices, symbols, _) => flush_dict(indices, symbols, nulls)?,
1779 Self::Duration(builder) => {
1780 let (_, vals, _) = builder.finish().into_parts();
1781 let vals = IntervalMonthDayNanoArray::try_new(vals, nulls)
1782 .map_err(|e| AvroError::ParseError(e.to_string()))?;
1783 Arc::new(vals)
1784 }
1785 #[cfg(feature = "avro_custom_types")]
1786 Self::RunEndEncoded(width, len, inner) => {
1787 let values = inner.flush(nulls)?;
1788 let n = *len;
1789 let arr = values.as_ref();
1790 let mut run_starts: Vec<usize> = Vec::with_capacity(n);
1791 if n > 0 {
1792 run_starts.push(0);
1793 for i in 1..n {
1794 if !values_equal_at(arr, i - 1, i) {
1795 run_starts.push(i);
1796 }
1797 }
1798 }
1799 if n > (u32::MAX as usize) {
1800 return Err(AvroError::InvalidArgument(format!(
1801 "RunEndEncoded length {n} exceeds maximum supported by UInt32 indices for take",
1802 )));
1803 }
1804 let run_count = run_starts.len();
1805 let take_idx: PrimitiveArray<UInt32Type> =
1806 run_starts.iter().map(|&s| s as u32).collect();
1807 let per_run_values = if run_count == 0 {
1808 values.slice(0, 0)
1809 } else {
1810 take(arr, &take_idx, Option::from(TakeOptions::default())).map_err(|e| {
1811 AvroError::ParseError(format!("take() for REE values failed: {e}"))
1812 })?
1813 };
1814
1815 macro_rules! build_run_array {
1816 ($Native:ty, $ArrowTy:ty) => {{
1817 let mut ends: Vec<$Native> = Vec::with_capacity(run_count);
1818 for (idx, &_start) in run_starts.iter().enumerate() {
1819 let end = if idx + 1 < run_count {
1820 run_starts[idx + 1]
1821 } else {
1822 n
1823 };
1824 ends.push(end as $Native);
1825 }
1826 let ends: PrimitiveArray<$ArrowTy> = ends.into_iter().collect();
1827 let run_arr = RunArray::<$ArrowTy>::try_new(&ends, per_run_values.as_ref())
1828 .map_err(|e| AvroError::ParseError(e.to_string()))?;
1829 Arc::new(run_arr) as ArrayRef
1830 }};
1831 }
1832 match *width {
1833 2 => {
1834 if n > i16::MAX as usize {
1835 return Err(AvroError::InvalidArgument(format!(
1836 "RunEndEncoded length {n} exceeds i16::MAX for run end width 2"
1837 )));
1838 }
1839 build_run_array!(i16, Int16Type)
1840 }
1841 4 => build_run_array!(i32, Int32Type),
1842 8 => build_run_array!(i64, Int64Type),
1843 other => {
1844 return Err(AvroError::InvalidArgument(format!(
1845 "Unsupported run-end width {other} for RunEndEncoded"
1846 )));
1847 }
1848 }
1849 }
1850 Self::Union(u) => u.flush(nulls)?,
1851 })
1852 }
1853}
1854
1855#[derive(Debug)]
1857enum NullablePlan {
1858 ReadTag {
1860 nullability: Nullability,
1861 resolution: ResolutionPlan,
1862 },
1863 FromSingle { resolution: ResolutionPlan },
1866}
1867
1868#[derive(Debug)]
1870enum ResolutionPlan {
1871 Promotion(Promotion),
1873 DefaultValue(AvroLiteral),
1875 EnumMapping(EnumResolution),
1877 Record(Projector),
1879}
1880
1881impl ResolutionPlan {
1882 fn try_new(decoder: &Decoder, resolution: &ResolutionInfo) -> Result<Self, AvroError> {
1883 match (decoder, resolution) {
1884 (_, ResolutionInfo::Promotion(p)) => Ok(ResolutionPlan::Promotion(*p)),
1885 (_, ResolutionInfo::DefaultValue(lit)) => Ok(ResolutionPlan::DefaultValue(lit.clone())),
1886 (_, ResolutionInfo::EnumMapping(m)) => {
1887 Ok(ResolutionPlan::EnumMapping(EnumResolution::new(m)))
1888 }
1889 (
1890 Decoder::Record {
1891 defaults: field_defaults,
1892 ..
1893 },
1894 ResolutionInfo::Record(r),
1895 ) => Ok(ResolutionPlan::Record(
1896 ProjectorBuilder::try_new(r, field_defaults).build()?,
1897 )),
1898 (_, ResolutionInfo::Record(_)) => Err(AvroError::SchemaError(
1899 "record resolution on non-record decoder".into(),
1900 )),
1901 (_, ResolutionInfo::Union(_)) => Err(AvroError::SchemaError(
1902 "union variant cannot be resolved to a union type".into(),
1903 )),
1904 }
1905 }
1906}
1907
1908#[derive(Debug)]
1909struct EnumResolution {
1910 mapping: Arc<[i32]>,
1911 default_index: i32,
1912}
1913
1914impl EnumResolution {
1915 fn new(mapping: &EnumMapping) -> Self {
1916 EnumResolution {
1917 mapping: mapping.mapping.clone(),
1918 default_index: mapping.default_index,
1919 }
1920 }
1921
1922 fn resolve(&self, index: i32) -> Result<i32, AvroError> {
1923 let resolved = usize::try_from(index)
1924 .ok()
1925 .and_then(|idx| self.mapping.get(idx).copied())
1926 .filter(|&idx| idx >= 0)
1927 .unwrap_or(self.default_index);
1928 if resolved >= 0 {
1929 Ok(resolved)
1930 } else {
1931 Err(AvroError::ParseError(format!(
1932 "Enum symbol index {index} not resolvable and no default provided",
1933 )))
1934 }
1935 }
1936}
1937
1938#[derive(Debug)]
1940struct DispatchLookupTable {
1941 to_reader: Box<[i8]>,
1957 resolution: Box<[ResolutionPlan]>,
1962}
1963
1964const NO_SOURCE: i8 = -1;
1967
1968impl DispatchLookupTable {
1969 fn from_writer_to_reader(
1970 reader_branches: &[Decoder],
1971 resolution_map: &[Option<(usize, ResolutionInfo)>],
1972 ) -> Result<Self, AvroError> {
1973 let mut to_reader = Vec::with_capacity(resolution_map.len());
1974 let mut resolution = Vec::with_capacity(resolution_map.len());
1975 for map in resolution_map {
1976 match map {
1977 Some((idx, res)) => {
1978 let idx = *idx;
1979 let idx_i8 = i8::try_from(idx).map_err(|_| {
1980 AvroError::SchemaError(format!(
1981 "Reader branch index {idx} exceeds i8 range (max {})",
1982 i8::MAX
1983 ))
1984 })?;
1985 let plan = ResolutionPlan::try_new(&reader_branches[idx], res)?;
1986 to_reader.push(idx_i8);
1987 resolution.push(plan);
1988 }
1989 None => {
1990 to_reader.push(NO_SOURCE);
1991 resolution.push(ResolutionPlan::DefaultValue(AvroLiteral::Null));
1992 }
1993 }
1994 }
1995 Ok(Self {
1996 to_reader: to_reader.into_boxed_slice(),
1997 resolution: resolution.into_boxed_slice(),
1998 })
1999 }
2000
2001 #[inline]
2003 fn resolve(&self, writer_index: usize) -> Option<(usize, &ResolutionPlan)> {
2004 let reader_index = *self.to_reader.get(writer_index)?;
2005 (reader_index >= 0).then(|| (reader_index as usize, &self.resolution[writer_index]))
2006 }
2007}
2008
2009#[derive(Debug)]
2010struct UnionDecoder {
2011 fields: UnionFields,
2012 branches: UnionDecoderBranches,
2013 default_emit_idx: usize,
2014 null_emit_idx: usize,
2015 plan: UnionReadPlan,
2016}
2017
2018#[derive(Debug, Default)]
2019struct UnionDecoderBranches {
2020 decoders: Vec<Decoder>,
2021 reader_type_codes: Vec<i8>,
2022 type_ids: Vec<i8>,
2023 offsets: Vec<i32>,
2024 counts: Vec<i32>,
2025}
2026
2027impl UnionDecoderBranches {
2028 fn new(decoders: Vec<Decoder>, reader_type_codes: Vec<i8>) -> Self {
2029 let branch_len = decoders.len().max(reader_type_codes.len());
2030 Self {
2031 decoders,
2032 reader_type_codes,
2033 type_ids: Vec::with_capacity(DEFAULT_CAPACITY),
2034 offsets: Vec::with_capacity(DEFAULT_CAPACITY),
2035 counts: vec![0; branch_len],
2036 }
2037 }
2038
2039 fn emit_to(&mut self, reader_idx: usize) -> Result<&mut Decoder, AvroError> {
2040 let branches_len = self.decoders.len();
2041 let Some(reader_branch) = self.decoders.get_mut(reader_idx) else {
2042 return Err(AvroError::ParseError(format!(
2043 "Union branch index {reader_idx} out of range ({branches_len} branches)"
2044 )));
2045 };
2046 self.type_ids.push(self.reader_type_codes[reader_idx]);
2047 self.offsets.push(self.counts[reader_idx]);
2048 self.counts[reader_idx] += 1;
2049 Ok(reader_branch)
2050 }
2051}
2052
2053impl Default for UnionDecoder {
2054 fn default() -> Self {
2055 Self {
2056 fields: UnionFields::empty(),
2057 branches: Default::default(),
2058 default_emit_idx: 0,
2059 null_emit_idx: 0,
2060 plan: UnionReadPlan::Passthrough,
2061 }
2062 }
2063}
2064
2065#[derive(Debug)]
2066enum UnionReadPlan {
2067 ReaderUnion {
2068 lookup_table: DispatchLookupTable,
2069 },
2070 FromSingle {
2071 reader_idx: usize,
2072 resolution: ResolutionPlan,
2073 },
2074 ToSingle {
2075 target: Box<Decoder>,
2076 lookup_table: DispatchLookupTable,
2077 },
2078 Passthrough,
2079}
2080
2081impl UnionReadPlan {
2082 fn from_resolved(
2083 reader_branches: &[Decoder],
2084 resolved: Option<ResolvedUnion>,
2085 ) -> Result<Self, AvroError> {
2086 let Some(info) = resolved else {
2087 return Ok(Self::Passthrough);
2088 };
2089 match (info.writer_is_union, info.reader_is_union) {
2090 (true, true) => {
2091 let lookup_table =
2092 DispatchLookupTable::from_writer_to_reader(reader_branches, &info.writer_to_reader)?;
2093 Ok(Self::ReaderUnion { lookup_table })
2094 }
2095 (false, true) => {
2096 let Some((idx, resolution)) =
2097 info.writer_to_reader.first().and_then(Option::as_ref)
2098 else {
2099 return Err(AvroError::SchemaError(
2100 "Writer type does not match any reader union branch".to_string(),
2101 ));
2102 };
2103 let reader_idx = *idx;
2104 Ok(Self::FromSingle {
2105 reader_idx,
2106 resolution: ResolutionPlan::try_new(&reader_branches[reader_idx], resolution)?,
2107 })
2108 }
2109 (true, false) => Err(AvroError::InvalidArgument(
2110 "UnionDecoder::try_new cannot build writer-union to single; use UnionDecoderBuilder with a target"
2111 .to_string(),
2112 )),
2113 _ => Err(AvroError::SchemaError(
2115 "ResolvedUnion constructed for non-union sides; resolver should return None"
2116 .to_string(),
2117 )),
2118 }
2119 }
2120}
2121
2122impl UnionDecoder {
2123 fn try_new(
2124 fields: UnionFields,
2125 branches: Vec<Decoder>,
2126 resolved: Option<ResolvedUnion>,
2127 ) -> Result<Self, AvroError> {
2128 let reader_type_codes = fields.iter().map(|(tid, _)| tid).collect::<Vec<i8>>();
2129 let null_branch = branches.iter().position(|b| matches!(b, Decoder::Null(_)));
2130 let default_emit_idx = 0;
2131 let null_emit_idx = null_branch.unwrap_or(default_emit_idx);
2132 let max_addr = (i32::MAX as usize) + 1;
2134 if branches.len() > max_addr {
2135 return Err(AvroError::SchemaError(format!(
2136 "Reader union has {} branches, which exceeds the maximum addressable \
2137 branches by an Avro int tag ({} + 1).",
2138 branches.len(),
2139 i32::MAX
2140 )));
2141 }
2142 let plan = UnionReadPlan::from_resolved(&branches, resolved)?;
2143 Ok(Self {
2144 fields,
2145 branches: UnionDecoderBranches::new(branches, reader_type_codes),
2146 default_emit_idx,
2147 null_emit_idx,
2148 plan,
2149 })
2150 }
2151
2152 fn with_single_target(target: Decoder, info: ResolvedUnion) -> Result<Self, AvroError> {
2153 debug_assert!(info.writer_is_union && !info.reader_is_union);
2155 let mut reader_branches = [target];
2156 let lookup_table =
2157 DispatchLookupTable::from_writer_to_reader(&reader_branches, &info.writer_to_reader)?;
2158 let target = Box::new(mem::replace(&mut reader_branches[0], Decoder::Null(0)));
2159 Ok(Self {
2160 plan: UnionReadPlan::ToSingle {
2161 target,
2162 lookup_table,
2163 },
2164 ..Self::default()
2165 })
2166 }
2167
2168 #[inline]
2169 fn read_tag(buf: &mut AvroCursor<'_>) -> Result<usize, AvroError> {
2170 let raw = buf.get_long()?;
2175 if raw < 0 {
2176 return Err(AvroError::ParseError(format!(
2177 "Negative union branch index {raw}"
2178 )));
2179 }
2180 usize::try_from(raw).map_err(|_| {
2181 AvroError::ParseError(format!(
2182 "Union branch index {raw} does not fit into usize on this platform ({}-bit)",
2183 (usize::BITS as usize)
2184 ))
2185 })
2186 }
2187
2188 #[inline]
2189 fn on_decoder<F>(&mut self, fallback_idx: usize, action: F) -> Result<(), AvroError>
2190 where
2191 F: FnOnce(&mut Decoder) -> Result<(), AvroError>,
2192 {
2193 if let UnionReadPlan::ToSingle { target, .. } = &mut self.plan {
2194 return action(target);
2195 }
2196 let reader_idx = match &self.plan {
2197 UnionReadPlan::FromSingle { reader_idx, .. } => *reader_idx,
2198 _ => fallback_idx,
2199 };
2200 self.branches.emit_to(reader_idx).and_then(action)
2201 }
2202
2203 fn append_null(&mut self) -> Result<(), AvroError> {
2204 self.on_decoder(self.null_emit_idx, |decoder| decoder.append_null())
2205 }
2206
2207 fn append_default(&mut self, lit: &AvroLiteral) -> Result<(), AvroError> {
2208 self.on_decoder(self.default_emit_idx, |decoder| decoder.append_default(lit))
2209 }
2210
2211 fn decode(&mut self, buf: &mut AvroCursor<'_>) -> Result<(), AvroError> {
2212 match &mut self.plan {
2213 UnionReadPlan::Passthrough => {
2214 let reader_idx = Self::read_tag(buf)?;
2215 let decoder = self.branches.emit_to(reader_idx)?;
2216 decoder.decode(buf)
2217 }
2218 UnionReadPlan::ReaderUnion { lookup_table } => {
2219 let idx = Self::read_tag(buf)?;
2220 let Some((reader_idx, resolution)) = lookup_table.resolve(idx) else {
2221 return Err(AvroError::ParseError(format!(
2222 "Union branch index {idx} not resolvable by reader schema"
2223 )));
2224 };
2225 let decoder = self.branches.emit_to(reader_idx)?;
2226 decoder.decode_with_resolution(buf, resolution)
2227 }
2228 UnionReadPlan::FromSingle {
2229 reader_idx,
2230 resolution,
2231 } => {
2232 let decoder = self.branches.emit_to(*reader_idx)?;
2233 decoder.decode_with_resolution(buf, resolution)
2234 }
2235 UnionReadPlan::ToSingle {
2236 target,
2237 lookup_table,
2238 } => {
2239 let idx = Self::read_tag(buf)?;
2240 let Some((_, resolution)) = lookup_table.resolve(idx) else {
2241 return Err(AvroError::ParseError(format!(
2242 "Writer union branch index {idx} not resolvable by reader schema"
2243 )));
2244 };
2245 target.decode_with_resolution(buf, resolution)
2246 }
2247 }
2248 }
2249
2250 fn flush(&mut self, nulls: Option<NullBuffer>) -> Result<ArrayRef, AvroError> {
2251 if let UnionReadPlan::ToSingle { target, .. } = &mut self.plan {
2252 return target.flush(nulls);
2253 }
2254 debug_assert!(
2255 nulls.is_none(),
2256 "UnionArray does not accept a validity bitmap; \
2257 nulls should have been materialized as a Null child during decode"
2258 );
2259 let children = self
2260 .branches
2261 .decoders
2262 .iter_mut()
2263 .map(|d| d.flush(None))
2264 .collect::<Result<Vec<_>, _>>()?;
2265 let arr = UnionArray::try_new(
2266 self.fields.clone(),
2267 flush_values(&mut self.branches.type_ids)
2268 .into_iter()
2269 .collect(),
2270 Some(
2271 flush_values(&mut self.branches.offsets)
2272 .into_iter()
2273 .collect(),
2274 ),
2275 children,
2276 )
2277 .map_err(|e| AvroError::ParseError(e.to_string()))?;
2278 self.branches.counts.fill(0);
2279 Ok(Arc::new(arr))
2280 }
2281}
2282
2283#[derive(Debug, Default)]
2284struct UnionDecoderBuilder {
2285 fields: Option<UnionFields>,
2286 branches: Option<Vec<Decoder>>,
2287 resolved: Option<ResolvedUnion>,
2288 target: Option<Decoder>,
2289}
2290
2291impl UnionDecoderBuilder {
2292 fn new() -> Self {
2293 Self::default()
2294 }
2295
2296 fn with_fields(mut self, fields: UnionFields) -> Self {
2297 self.fields = Some(fields);
2298 self
2299 }
2300
2301 fn with_branches(mut self, branches: Vec<Decoder>) -> Self {
2302 self.branches = Some(branches);
2303 self
2304 }
2305
2306 fn with_resolved_union(mut self, resolved_union: ResolvedUnion) -> Self {
2307 self.resolved = Some(resolved_union);
2308 self
2309 }
2310
2311 fn with_target(mut self, target: Decoder) -> Self {
2312 self.target = Some(target);
2313 self
2314 }
2315
2316 fn build(self) -> Result<UnionDecoder, AvroError> {
2317 match (self.resolved, self.fields, self.branches, self.target) {
2318 (resolved, Some(fields), Some(branches), None) => {
2319 UnionDecoder::try_new(fields, branches, resolved)
2320 }
2321 (Some(info), None, None, Some(target))
2322 if info.writer_is_union && !info.reader_is_union =>
2323 {
2324 UnionDecoder::with_single_target(target, info)
2325 }
2326 _ => Err(AvroError::InvalidArgument(
2327 "Invalid UnionDecoderBuilder configuration: expected either \
2328 (fields + branches + resolved) with no target for reader-unions, or \
2329 (resolved + target) with no fields/branches for writer-union to single."
2330 .to_string(),
2331 )),
2332 }
2333 }
2334}
2335
2336#[derive(Debug, Copy, Clone)]
2337enum NegativeBlockBehavior {
2338 ProcessItems,
2339 SkipBySize,
2340}
2341
2342#[inline]
2343fn skip_blocks(
2344 buf: &mut AvroCursor,
2345 mut skip_item: impl FnMut(&mut AvroCursor) -> Result<(), AvroError>,
2346) -> Result<usize, AvroError> {
2347 process_blockwise(
2348 buf,
2349 move |c| skip_item(c),
2350 NegativeBlockBehavior::SkipBySize,
2351 )
2352}
2353
2354#[inline]
2355fn flush_dict(
2356 indices: &mut Vec<i32>,
2357 symbols: &[String],
2358 nulls: Option<NullBuffer>,
2359) -> Result<ArrayRef, AvroError> {
2360 let keys = flush_primitive::<Int32Type>(indices, nulls);
2361 let values = Arc::new(StringArray::from_iter_values(
2362 symbols.iter().map(|s| s.as_str()),
2363 ));
2364 DictionaryArray::try_new(keys, values)
2365 .map_err(Into::into)
2366 .map(|arr| Arc::new(arr) as ArrayRef)
2367}
2368
2369#[inline]
2370fn read_blocks(
2371 buf: &mut AvroCursor,
2372 decode_entry: impl FnMut(&mut AvroCursor) -> Result<(), AvroError>,
2373) -> Result<usize, AvroError> {
2374 process_blockwise(buf, decode_entry, NegativeBlockBehavior::ProcessItems)
2375}
2376
2377#[inline]
2378fn process_blockwise(
2379 buf: &mut AvroCursor,
2380 mut on_item: impl FnMut(&mut AvroCursor) -> Result<(), AvroError>,
2381 negative_behavior: NegativeBlockBehavior,
2382) -> Result<usize, AvroError> {
2383 let mut total = 0usize;
2384 loop {
2385 let block_count = buf.get_long()?;
2390 match block_count.cmp(&0) {
2391 Ordering::Equal => break,
2392 Ordering::Less => {
2393 let count = block_count.unsigned_abs() as usize;
2395 let raw_size = buf.get_long()?;
2397 let size_in_bytes = usize::try_from(raw_size).map_err(|_| {
2398 AvroError::ParseError(format!("Block size cannot be negative, got {raw_size}"))
2399 })?;
2400 match negative_behavior {
2401 NegativeBlockBehavior::ProcessItems => {
2402 total = process_block_items(buf, count, total, &mut on_item)?;
2404 }
2405 NegativeBlockBehavior::SkipBySize => {
2406 let _ = buf.get_fixed(size_in_bytes)?;
2408 total = total.saturating_add(count);
2409 }
2410 }
2411 }
2412 Ordering::Greater => {
2413 let count = block_count as usize;
2414 total = process_block_items(buf, count, total, &mut on_item)?;
2415 }
2416 }
2417 }
2418 Ok(total)
2419}
2420
2421#[inline]
2426fn process_block_items(
2427 buf: &mut AvroCursor,
2428 count: usize,
2429 total: usize,
2430 on_item: &mut impl FnMut(&mut AvroCursor) -> Result<(), AvroError>,
2431) -> Result<usize, AvroError> {
2432 let Some(new_total) = total
2433 .checked_add(count)
2434 .filter(|&t| i32::try_from(t).is_ok())
2435 else {
2436 return Err(AvroError::ParseError(
2437 "Capacity overflow when decoding array/map item blocks".to_string(),
2438 ));
2439 };
2440 for _ in 0..count {
2441 on_item(buf)?;
2442 }
2443 Ok(new_total)
2444}
2445
2446#[inline]
2447fn flush_values<T>(values: &mut Vec<T>) -> Vec<T> {
2448 std::mem::replace(values, Vec::with_capacity(DEFAULT_CAPACITY))
2449}
2450
2451#[inline]
2452fn flush_offsets(offsets: &mut OffsetBufferBuilder<i32>) -> OffsetBuffer<i32> {
2453 std::mem::replace(offsets, OffsetBufferBuilder::new(DEFAULT_CAPACITY)).finish()
2454}
2455
2456#[inline]
2457fn flush_primitive<T: ArrowPrimitiveType>(
2458 values: &mut Vec<T::Native>,
2459 nulls: Option<NullBuffer>,
2460) -> PrimitiveArray<T> {
2461 PrimitiveArray::new(flush_values(values).into(), nulls)
2462}
2463
2464#[inline]
2465fn read_decimal_bytes_be<const N: usize>(
2466 buf: &mut AvroCursor<'_>,
2467 size: Option<usize>,
2468) -> Result<[u8; N], AvroError> {
2469 match size {
2470 Some(n) if n == N => {
2471 let raw = buf.get_fixed(N)?;
2472 let mut arr = [0u8; N];
2473 arr.copy_from_slice(raw);
2474 Ok(arr)
2475 }
2476 Some(n) => {
2477 let raw = buf.get_fixed(n)?;
2478 sign_cast_to::<N>(raw)
2479 }
2480 None => {
2481 let raw = buf.get_bytes()?;
2482 sign_cast_to::<N>(raw)
2483 }
2484 }
2485}
2486
2487#[inline]
2496fn sign_cast_to<const N: usize>(raw: &[u8]) -> Result<[u8; N], AvroError> {
2497 let len = raw.len();
2498 if len == N {
2500 let mut out = [0u8; N];
2501 out.copy_from_slice(raw);
2502 return Ok(out);
2503 }
2504 let first = raw.first().copied().unwrap_or(0u8);
2506 let sign_byte = if (first & 0x80) == 0 { 0x00 } else { 0xFF };
2507 let mut out = [sign_byte; N];
2509 if len > N {
2510 let extra = len - N;
2513 if raw[..extra].iter().any(|&b| b != sign_byte) {
2515 return Err(AvroError::ParseError(format!(
2516 "Decimal value with {len} bytes cannot be represented in {N} bytes without overflow"
2517 )));
2518 }
2519 if N > 0 {
2520 let first_kept = raw[extra];
2521 let sign_bit_mismatch = ((first_kept ^ sign_byte) & 0x80) != 0;
2522 if sign_bit_mismatch {
2523 return Err(AvroError::ParseError(format!(
2524 "Decimal value with {len} bytes cannot be represented in {N} bytes without overflow"
2525 )));
2526 }
2527 }
2528 out.copy_from_slice(&raw[extra..]);
2529 return Ok(out);
2530 }
2531 out[N - len..].copy_from_slice(raw);
2532 Ok(out)
2533}
2534
2535#[cfg(feature = "avro_custom_types")]
2536#[inline]
2537fn values_equal_at(arr: &dyn Array, i: usize, j: usize) -> bool {
2538 match (arr.is_null(i), arr.is_null(j)) {
2539 (true, true) => true,
2540 (true, false) | (false, true) => false,
2541 (false, false) => {
2542 let a = arr.slice(i, 1);
2543 let b = arr.slice(j, 1);
2544 a == b
2545 }
2546 }
2547}
2548
2549#[derive(Debug)]
2550struct Projector {
2551 writer_projections: Vec<FieldProjection>,
2552 default_injections: Arc<[(usize, AvroLiteral)]>,
2553}
2554
2555#[derive(Debug)]
2556enum FieldProjection {
2557 ToReader(usize),
2558 Skip(Skipper),
2559}
2560
2561#[derive(Debug)]
2562struct ProjectorBuilder<'a> {
2563 rec: &'a ResolvedRecord,
2564 field_defaults: &'a [Option<AvroLiteral>],
2565}
2566
2567impl<'a> ProjectorBuilder<'a> {
2568 #[inline]
2569 fn try_new(rec: &'a ResolvedRecord, field_defaults: &'a [Option<AvroLiteral>]) -> Self {
2570 Self {
2571 rec,
2572 field_defaults,
2573 }
2574 }
2575
2576 #[inline]
2577 fn build(self) -> Result<Projector, AvroError> {
2578 let mut default_injections: Vec<(usize, AvroLiteral)> =
2579 Vec::with_capacity(self.rec.default_fields.len());
2580 for &idx in self.rec.default_fields.as_ref() {
2581 let lit = self
2582 .field_defaults
2583 .get(idx)
2584 .and_then(|lit| lit.clone())
2585 .unwrap_or(AvroLiteral::Null);
2586 default_injections.push((idx, lit));
2587 }
2588 let writer_projections = self
2589 .rec
2590 .writer_fields
2591 .iter()
2592 .map(|field| match field {
2593 ResolvedField::ToReader(index, _) => Ok(FieldProjection::ToReader(*index)),
2594 ResolvedField::Skip(datatype) => {
2595 let skipper = Skipper::from_avro(datatype)?;
2596 Ok(FieldProjection::Skip(skipper))
2597 }
2598 })
2599 .collect::<Result<_, AvroError>>()?;
2600 Ok(Projector {
2601 writer_projections,
2602 default_injections: default_injections.into(),
2603 })
2604 }
2605}
2606
2607impl Projector {
2608 #[inline]
2609 fn project_record(
2610 &self,
2611 buf: &mut AvroCursor<'_>,
2612 encodings: &mut [Decoder],
2613 ) -> Result<(), AvroError> {
2614 for field_proj in &self.writer_projections {
2615 match field_proj {
2616 FieldProjection::ToReader(index) => encodings[*index].decode(buf)?,
2617 FieldProjection::Skip(skipper) => skipper.skip(buf)?,
2618 }
2619 }
2620 for (reader_index, lit) in self.default_injections.as_ref() {
2621 encodings[*reader_index].append_default(lit)?;
2622 }
2623 Ok(())
2624 }
2625}
2626
2627#[derive(Debug)]
2633enum Skipper {
2634 Null,
2635 Boolean,
2636 Int32,
2637 Int64,
2638 Float32,
2639 Float64,
2640 Bytes,
2641 String,
2642 TimeMicros,
2643 TimestampMillis,
2644 TimestampMicros,
2645 TimestampNanos,
2646 Fixed(usize),
2647 Decimal(Option<usize>),
2648 UuidString,
2649 Enum,
2650 DurationFixed12,
2651 List(Box<Skipper>),
2652 Map(Box<Skipper>),
2653 Struct(Vec<Skipper>),
2654 Union(Vec<Skipper>),
2655 Nullable(Nullability, Box<Skipper>),
2656 #[cfg(feature = "avro_custom_types")]
2657 RunEndEncoded(Box<Skipper>),
2658}
2659
2660impl Skipper {
2661 fn from_avro(dt: &AvroDataType) -> Result<Self, AvroError> {
2662 let mut base = match dt.codec() {
2663 Codec::Null => Self::Null,
2664 Codec::Boolean => Self::Boolean,
2665 Codec::Int32 | Codec::Date32 | Codec::TimeMillis => Self::Int32,
2666 Codec::Int64 => Self::Int64,
2667 Codec::TimeMicros => Self::TimeMicros,
2668 Codec::TimestampMillis(_) => Self::TimestampMillis,
2669 Codec::TimestampMicros(_) => Self::TimestampMicros,
2670 Codec::TimestampNanos(_) => Self::TimestampNanos,
2671 #[cfg(feature = "avro_custom_types")]
2672 Codec::DurationNanos
2673 | Codec::DurationMicros
2674 | Codec::DurationMillis
2675 | Codec::DurationSeconds => Self::Int64,
2676 #[cfg(feature = "avro_custom_types")]
2677 Codec::Int8 | Codec::Int16 | Codec::UInt8 | Codec::UInt16 | Codec::Time32Secs => {
2678 Self::Int32
2679 }
2680 #[cfg(feature = "avro_custom_types")]
2681 Codec::UInt32 | Codec::Date64 | Codec::TimeNanos | Codec::TimestampSecs(_) => {
2682 Self::Int64
2683 }
2684 #[cfg(feature = "avro_custom_types")]
2685 Codec::UInt64 => Self::Fixed(8),
2686 #[cfg(feature = "avro_custom_types")]
2687 Codec::Float16 => Self::Fixed(2),
2688 #[cfg(feature = "avro_custom_types")]
2689 Codec::IntervalYearMonth => Self::Fixed(4),
2690 #[cfg(feature = "avro_custom_types")]
2691 Codec::IntervalMonthDayNano => Self::Fixed(16),
2692 #[cfg(feature = "avro_custom_types")]
2693 Codec::IntervalDayTime => Self::Fixed(8),
2694 Codec::Float32 => Self::Float32,
2695 Codec::Float64 => Self::Float64,
2696 Codec::Binary => Self::Bytes,
2697 Codec::Utf8 | Codec::Utf8View => Self::String,
2698 Codec::Fixed(sz) => Self::Fixed(*sz as usize),
2699 Codec::Decimal(_, _, size) => Self::Decimal(*size),
2700 Codec::Uuid => Self::UuidString, Codec::Enum(_) => Self::Enum,
2702 Codec::List(item) => Self::List(Box::new(Skipper::from_avro(item)?)),
2703 Codec::Struct(fields) => {
2704 if let Some(ResolutionInfo::Record(rec)) = dt.resolution.as_ref() {
2705 Self::Struct(
2706 rec.writer_fields
2707 .iter()
2708 .map(|wf| match wf {
2709 ResolvedField::ToReader(_, wdt) | ResolvedField::Skip(wdt) => {
2710 Skipper::from_avro(wdt)
2711 }
2712 })
2713 .collect::<Result<_, _>>()?,
2714 )
2715 } else {
2716 Self::Struct(
2717 fields
2718 .iter()
2719 .map(|f| Skipper::from_avro(f.data_type()))
2720 .collect::<Result<_, _>>()?,
2721 )
2722 }
2723 }
2724 Codec::Map(values) => Self::Map(Box::new(Skipper::from_avro(values)?)),
2725 Codec::Interval => Self::DurationFixed12,
2726 Codec::Union(encodings, _, _) => {
2727 let max_addr = (i32::MAX as usize) + 1;
2728 if encodings.len() > max_addr {
2729 return Err(AvroError::SchemaError(format!(
2730 "Writer union has {} branches, which exceeds the maximum addressable \
2731 branches by an Avro int tag ({} + 1).",
2732 encodings.len(),
2733 i32::MAX
2734 )));
2735 }
2736 Self::Union(
2737 encodings
2738 .iter()
2739 .map(Skipper::from_avro)
2740 .collect::<Result<_, _>>()?,
2741 )
2742 }
2743 #[cfg(feature = "avro_custom_types")]
2744 Codec::RunEndEncoded(inner, _w) => {
2745 Self::RunEndEncoded(Box::new(Skipper::from_avro(inner)?))
2746 }
2747 };
2748 if let Some(n) = dt.nullability() {
2749 base = Self::Nullable(n, Box::new(base));
2750 }
2751 Ok(base)
2752 }
2753
2754 fn skip(&self, buf: &mut AvroCursor<'_>) -> Result<(), AvroError> {
2755 match self {
2756 Self::Null => Ok(()),
2757 Self::Boolean => {
2758 buf.get_bool()?;
2759 Ok(())
2760 }
2761 Self::Int32 => {
2762 buf.skip_int()?;
2763 Ok(())
2764 }
2765 Self::Int64
2766 | Self::TimeMicros
2767 | Self::TimestampMillis
2768 | Self::TimestampMicros
2769 | Self::TimestampNanos => {
2770 buf.skip_long()?;
2771 Ok(())
2772 }
2773 Self::Float32 => {
2774 buf.get_float()?;
2775 Ok(())
2776 }
2777 Self::Float64 => {
2778 buf.get_double()?;
2779 Ok(())
2780 }
2781 Self::Bytes | Self::String | Self::UuidString => {
2782 buf.get_bytes()?;
2783 Ok(())
2784 }
2785 Self::Fixed(sz) => {
2786 buf.get_fixed(*sz)?;
2787 Ok(())
2788 }
2789 Self::Decimal(size) => {
2790 if let Some(s) = size {
2791 buf.get_fixed(*s)
2792 } else {
2793 buf.get_bytes()
2794 }?;
2795 Ok(())
2796 }
2797 Self::Enum => {
2798 buf.skip_int()?;
2799 Ok(())
2800 }
2801 Self::DurationFixed12 => {
2802 buf.get_fixed(12)?;
2803 Ok(())
2804 }
2805 Self::List(item) => {
2806 skip_blocks(buf, |c| item.skip(c))?;
2807 Ok(())
2808 }
2809 Self::Map(value) => {
2810 skip_blocks(buf, |c| {
2811 c.get_bytes()?; value.skip(c)
2813 })?;
2814 Ok(())
2815 }
2816 Self::Struct(fields) => {
2817 for f in fields {
2818 f.skip(buf)?
2819 }
2820 Ok(())
2821 }
2822 Self::Union(encodings) => {
2823 let raw = buf.get_long()?;
2825 if raw < 0 {
2826 return Err(AvroError::ParseError(format!(
2827 "Negative union branch index {raw}"
2828 )));
2829 }
2830 let idx: usize = usize::try_from(raw).map_err(|_| {
2831 AvroError::ParseError(format!(
2832 "Union branch index {raw} does not fit into usize on this platform ({}-bit)",
2833 (usize::BITS as usize)
2834 ))
2835 })?;
2836 let Some(encoding) = encodings.get(idx) else {
2837 return Err(AvroError::ParseError(format!(
2838 "Union branch index {idx} out of range for skipper ({} branches)",
2839 encodings.len()
2840 )));
2841 };
2842 encoding.skip(buf)
2843 }
2844 Self::Nullable(order, inner) => {
2845 let branch = buf.read_vlq()?;
2846 let is_not_null = match *order {
2847 Nullability::NullFirst => branch != 0,
2848 Nullability::NullSecond => branch == 0,
2849 };
2850 if is_not_null {
2851 inner.skip(buf)?;
2852 }
2853 Ok(())
2854 }
2855 #[cfg(feature = "avro_custom_types")]
2856 Self::RunEndEncoded(inner) => inner.skip(buf),
2857 }
2858 }
2859}
2860
2861#[cfg(test)]
2862mod tests {
2863 use super::*;
2864 use crate::codec::AvroFieldBuilder;
2865 use crate::schema::{Attributes, ComplexType, Field, PrimitiveType, Record, Schema, TypeName};
2866 use arrow_array::cast::AsArray;
2867 use indexmap::IndexMap;
2868 use std::collections::HashMap;
2869
2870 fn encode_avro_int(value: i32) -> Vec<u8> {
2871 let mut buf = Vec::new();
2872 let mut v = (value << 1) ^ (value >> 31);
2873 while v & !0x7F != 0 {
2874 buf.push(((v & 0x7F) | 0x80) as u8);
2875 v >>= 7;
2876 }
2877 buf.push(v as u8);
2878 buf
2879 }
2880
2881 fn encode_avro_long(value: i64) -> Vec<u8> {
2882 let mut buf = Vec::new();
2883 let mut v = (value << 1) ^ (value >> 63);
2884 while v & !0x7F != 0 {
2885 buf.push(((v & 0x7F) | 0x80) as u8);
2886 v >>= 7;
2887 }
2888 buf.push(v as u8);
2889 buf
2890 }
2891
2892 fn encode_avro_bytes(bytes: &[u8]) -> Vec<u8> {
2893 let mut buf = encode_avro_long(bytes.len() as i64);
2894 buf.extend_from_slice(bytes);
2895 buf
2896 }
2897
2898 fn avro_from_codec(codec: Codec) -> AvroDataType {
2899 AvroDataType::new(codec, Default::default(), None)
2900 }
2901
2902 fn resolved_root_datatype(
2903 writer: Schema<'static>,
2904 reader: Schema<'static>,
2905 use_utf8view: bool,
2906 strict_mode: bool,
2907 ) -> AvroDataType {
2908 let writer_record = Schema::Complex(ComplexType::Record(Record {
2910 name: "Root",
2911 namespace: None,
2912 doc: None,
2913 aliases: vec![],
2914 fields: vec![Field {
2915 name: "v",
2916 r#type: writer,
2917 default: None,
2918 doc: None,
2919 aliases: vec![],
2920 }],
2921 attributes: Attributes::default(),
2922 }));
2923
2924 let reader_record = Schema::Complex(ComplexType::Record(Record {
2926 name: "Root",
2927 namespace: None,
2928 doc: None,
2929 aliases: vec![],
2930 fields: vec![Field {
2931 name: "v",
2932 r#type: reader,
2933 default: None,
2934 doc: None,
2935 aliases: vec![],
2936 }],
2937 attributes: Attributes::default(),
2938 }));
2939
2940 let field = AvroFieldBuilder::new(&writer_record)
2942 .with_reader_schema(&reader_record)
2943 .with_utf8view(use_utf8view)
2944 .with_strict_mode(strict_mode)
2945 .build()
2946 .expect("schema resolution should succeed");
2947
2948 match field.data_type().codec() {
2949 Codec::Struct(fields) => fields[0].data_type().clone(),
2950 other => panic!("expected wrapper struct, got {other:?}"),
2951 }
2952 }
2953
2954 fn decoder_for_promotion(
2955 writer: PrimitiveType,
2956 reader: PrimitiveType,
2957 use_utf8view: bool,
2958 ) -> Decoder {
2959 let ws = Schema::TypeName(TypeName::Primitive(writer));
2960 let rs = Schema::TypeName(TypeName::Primitive(reader));
2961 let dt = resolved_root_datatype(ws, rs, use_utf8view, false);
2962 Decoder::try_new(&dt).unwrap()
2963 }
2964
2965 fn make_avro_dt(codec: Codec, nullability: Option<Nullability>) -> AvroDataType {
2966 AvroDataType::new(codec, HashMap::new(), nullability)
2967 }
2968
2969 #[cfg(feature = "avro_custom_types")]
2970 fn encode_vlq_u64(mut x: u64) -> Vec<u8> {
2971 let mut out = Vec::with_capacity(10);
2972 while x >= 0x80 {
2973 out.push((x as u8) | 0x80);
2974 x >>= 7;
2975 }
2976 out.push(x as u8);
2977 out
2978 }
2979
2980 #[test]
2981 fn test_union_resolution_writer_union_reader_union_reorder_and_promotion_dense() {
2982 let ws = Schema::Union(vec![
2983 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2984 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2985 ]);
2986 let rs = Schema::Union(vec![
2987 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2988 Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
2989 ]);
2990
2991 let dt = resolved_root_datatype(ws, rs, false, false);
2992 let mut dec = Decoder::try_new(&dt).unwrap();
2993
2994 let mut rec1 = encode_avro_long(0);
2995 rec1.extend(encode_avro_int(7));
2996 let mut cur1 = AvroCursor::new(&rec1);
2997 dec.decode(&mut cur1).unwrap();
2998
2999 let mut rec2 = encode_avro_long(1);
3000 rec2.extend(encode_avro_bytes(b"abc"));
3001 let mut cur2 = AvroCursor::new(&rec2);
3002 dec.decode(&mut cur2).unwrap();
3003
3004 let arr = dec.flush(None).unwrap();
3005 let ua = arr
3006 .as_any()
3007 .downcast_ref::<UnionArray>()
3008 .expect("dense union output");
3009
3010 assert_eq!(
3011 ua.type_id(0),
3012 1,
3013 "first value must select reader 'long' branch"
3014 );
3015 assert_eq!(ua.value_offset(0), 0);
3016
3017 assert_eq!(
3018 ua.type_id(1),
3019 0,
3020 "second value must select reader 'string' branch"
3021 );
3022 assert_eq!(ua.value_offset(1), 0);
3023
3024 let long_child = ua.child(1).as_any().downcast_ref::<Int64Array>().unwrap();
3025 assert_eq!(long_child.len(), 1);
3026 assert_eq!(long_child.value(0), 7);
3027
3028 let str_child = ua.child(0).as_any().downcast_ref::<StringArray>().unwrap();
3029 assert_eq!(str_child.len(), 1);
3030 assert_eq!(str_child.value(0), "abc");
3031 }
3032
3033 #[test]
3034 fn test_union_resolution_writer_union_reader_nonunion_promotion_int_to_long() {
3035 let ws = Schema::Union(vec![
3036 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3037 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3038 ]);
3039 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Long));
3040
3041 let dt = resolved_root_datatype(ws, rs, false, false);
3042 let mut dec = Decoder::try_new(&dt).unwrap();
3043
3044 let mut data = encode_avro_long(0);
3045 data.extend(encode_avro_int(5));
3046 let mut cur = AvroCursor::new(&data);
3047 dec.decode(&mut cur).unwrap();
3048
3049 let arr = dec.flush(None).unwrap();
3050 let out = arr.as_any().downcast_ref::<Int64Array>().unwrap();
3051 assert_eq!(out.len(), 1);
3052 assert_eq!(out.value(0), 5);
3053 }
3054
3055 #[test]
3056 fn test_union_resolution_writer_union_reader_nonunion_mismatch_errors() {
3057 let ws = Schema::Union(vec![
3058 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3059 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3060 ]);
3061 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Long));
3062
3063 let dt = resolved_root_datatype(ws, rs, false, false);
3064 let mut dec = Decoder::try_new(&dt).unwrap();
3065
3066 let mut data = encode_avro_long(1);
3067 data.extend(encode_avro_bytes(b"z"));
3068 let mut cur = AvroCursor::new(&data);
3069 let res = dec.decode(&mut cur);
3070 assert!(
3071 res.is_err(),
3072 "expected error when writer union branch does not resolve to reader non-union type"
3073 );
3074 }
3075
3076 #[test]
3077 fn test_union_resolution_writer_nonunion_reader_union_selects_matching_branch() {
3078 let ws = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3079 let rs = Schema::Union(vec![
3080 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3081 Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3082 ]);
3083
3084 let dt = resolved_root_datatype(ws, rs, false, false);
3085 let mut dec = Decoder::try_new(&dt).unwrap();
3086
3087 let data = encode_avro_int(6);
3088 let mut cur = AvroCursor::new(&data);
3089 dec.decode(&mut cur).unwrap();
3090
3091 let arr = dec.flush(None).unwrap();
3092 let ua = arr
3093 .as_any()
3094 .downcast_ref::<UnionArray>()
3095 .expect("dense union output");
3096 assert_eq!(ua.len(), 1);
3097 assert_eq!(
3098 ua.type_id(0),
3099 1,
3100 "must resolve to reader 'long' branch (type_id 1)"
3101 );
3102 assert_eq!(ua.value_offset(0), 0);
3103
3104 let long_child = ua.child(1).as_any().downcast_ref::<Int64Array>().unwrap();
3105 assert_eq!(long_child.len(), 1);
3106 assert_eq!(long_child.value(0), 6);
3107
3108 let str_child = ua.child(0).as_any().downcast_ref::<StringArray>().unwrap();
3109 assert_eq!(str_child.len(), 0, "string branch must be empty");
3110 }
3111
3112 #[test]
3113 fn test_union_resolution_writer_union_reader_union_unmapped_branch_errors() {
3114 let ws = Schema::Union(vec![
3115 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3116 Schema::TypeName(TypeName::Primitive(PrimitiveType::Boolean)),
3117 ]);
3118 let rs = Schema::Union(vec![
3119 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3120 Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3121 ]);
3122
3123 let dt = resolved_root_datatype(ws, rs, false, false);
3124 let mut dec = Decoder::try_new(&dt).unwrap();
3125
3126 let mut data = encode_avro_long(1);
3127 data.push(1);
3128 let mut cur = AvroCursor::new(&data);
3129 let res = dec.decode(&mut cur);
3130 assert!(
3131 res.is_err(),
3132 "expected error for unmapped writer 'boolean' branch"
3133 );
3134 }
3135
3136 #[test]
3137 fn test_schema_resolution_promotion_int_to_long() {
3138 let mut dec = decoder_for_promotion(PrimitiveType::Int, PrimitiveType::Long, false);
3139 assert!(matches!(dec, Decoder::Int32ToInt64(_)));
3140 for v in [0, 1, -2, 123456] {
3141 let data = encode_avro_int(v);
3142 let mut cur = AvroCursor::new(&data);
3143 dec.decode(&mut cur).unwrap();
3144 }
3145 let arr = dec.flush(None).unwrap();
3146 let a = arr.as_any().downcast_ref::<Int64Array>().unwrap();
3147 assert_eq!(a.value(0), 0);
3148 assert_eq!(a.value(1), 1);
3149 assert_eq!(a.value(2), -2);
3150 assert_eq!(a.value(3), 123456);
3151 }
3152
3153 #[test]
3154 fn test_schema_resolution_promotion_int_to_float() {
3155 let mut dec = decoder_for_promotion(PrimitiveType::Int, PrimitiveType::Float, false);
3156 assert!(matches!(dec, Decoder::Int32ToFloat32(_)));
3157 for v in [0, 42, -7] {
3158 let data = encode_avro_int(v);
3159 let mut cur = AvroCursor::new(&data);
3160 dec.decode(&mut cur).unwrap();
3161 }
3162 let arr = dec.flush(None).unwrap();
3163 let a = arr.as_any().downcast_ref::<Float32Array>().unwrap();
3164 assert_eq!(a.value(0), 0.0);
3165 assert_eq!(a.value(1), 42.0);
3166 assert_eq!(a.value(2), -7.0);
3167 }
3168
3169 #[test]
3170 fn test_schema_resolution_promotion_int_to_double() {
3171 let mut dec = decoder_for_promotion(PrimitiveType::Int, PrimitiveType::Double, false);
3172 assert!(matches!(dec, Decoder::Int32ToFloat64(_)));
3173 for v in [1, -1, 10_000] {
3174 let data = encode_avro_int(v);
3175 let mut cur = AvroCursor::new(&data);
3176 dec.decode(&mut cur).unwrap();
3177 }
3178 let arr = dec.flush(None).unwrap();
3179 let a = arr.as_any().downcast_ref::<Float64Array>().unwrap();
3180 assert_eq!(a.value(0), 1.0);
3181 assert_eq!(a.value(1), -1.0);
3182 assert_eq!(a.value(2), 10_000.0);
3183 }
3184
3185 #[test]
3186 fn test_schema_resolution_promotion_long_to_float() {
3187 let mut dec = decoder_for_promotion(PrimitiveType::Long, PrimitiveType::Float, false);
3188 assert!(matches!(dec, Decoder::Int64ToFloat32(_)));
3189 for v in [0_i64, 1_000_000_i64, -123_i64] {
3190 let data = encode_avro_long(v);
3191 let mut cur = AvroCursor::new(&data);
3192 dec.decode(&mut cur).unwrap();
3193 }
3194 let arr = dec.flush(None).unwrap();
3195 let a = arr.as_any().downcast_ref::<Float32Array>().unwrap();
3196 assert_eq!(a.value(0), 0.0);
3197 assert_eq!(a.value(1), 1_000_000.0);
3198 assert_eq!(a.value(2), -123.0);
3199 }
3200
3201 #[test]
3202 fn test_schema_resolution_promotion_long_to_double() {
3203 let mut dec = decoder_for_promotion(PrimitiveType::Long, PrimitiveType::Double, false);
3204 assert!(matches!(dec, Decoder::Int64ToFloat64(_)));
3205 for v in [2_i64, -2_i64, 9_223_372_i64] {
3206 let data = encode_avro_long(v);
3207 let mut cur = AvroCursor::new(&data);
3208 dec.decode(&mut cur).unwrap();
3209 }
3210 let arr = dec.flush(None).unwrap();
3211 let a = arr.as_any().downcast_ref::<Float64Array>().unwrap();
3212 assert_eq!(a.value(0), 2.0);
3213 assert_eq!(a.value(1), -2.0);
3214 assert_eq!(a.value(2), 9_223_372.0);
3215 }
3216
3217 #[test]
3218 fn test_schema_resolution_promotion_float_to_double() {
3219 let mut dec = decoder_for_promotion(PrimitiveType::Float, PrimitiveType::Double, false);
3220 assert!(matches!(dec, Decoder::Float32ToFloat64(_)));
3221 for v in [0.5_f32, -3.25_f32, 1.0e6_f32] {
3222 let data = v.to_le_bytes().to_vec();
3223 let mut cur = AvroCursor::new(&data);
3224 dec.decode(&mut cur).unwrap();
3225 }
3226 let arr = dec.flush(None).unwrap();
3227 let a = arr.as_any().downcast_ref::<Float64Array>().unwrap();
3228 assert_eq!(a.value(0), 0.5_f64);
3229 assert_eq!(a.value(1), -3.25_f64);
3230 assert_eq!(a.value(2), 1.0e6_f64);
3231 }
3232
3233 #[test]
3234 fn test_schema_resolution_promotion_bytes_to_string_utf8() {
3235 let mut dec = decoder_for_promotion(PrimitiveType::Bytes, PrimitiveType::String, false);
3236 assert!(matches!(dec, Decoder::BytesToString(_, _)));
3237 for s in ["hello", "world", "héllo"] {
3238 let data = encode_avro_bytes(s.as_bytes());
3239 let mut cur = AvroCursor::new(&data);
3240 dec.decode(&mut cur).unwrap();
3241 }
3242 let arr = dec.flush(None).unwrap();
3243 let a = arr.as_any().downcast_ref::<StringArray>().unwrap();
3244 assert_eq!(a.value(0), "hello");
3245 assert_eq!(a.value(1), "world");
3246 assert_eq!(a.value(2), "héllo");
3247 }
3248
3249 #[test]
3250 fn test_schema_resolution_promotion_bytes_to_string_utf8view_enabled() {
3251 let mut dec = decoder_for_promotion(PrimitiveType::Bytes, PrimitiveType::String, true);
3252 assert!(matches!(dec, Decoder::BytesToString(_, _)));
3253 let data = encode_avro_bytes(b"abc");
3254 let mut cur = AvroCursor::new(&data);
3255 dec.decode(&mut cur).unwrap();
3256 let arr = dec.flush(None).unwrap();
3257 let a = arr.as_any().downcast_ref::<StringArray>().unwrap();
3258 assert_eq!(a.value(0), "abc");
3259 }
3260
3261 #[test]
3262 fn test_schema_resolution_promotion_string_to_bytes() {
3263 let mut dec = decoder_for_promotion(PrimitiveType::String, PrimitiveType::Bytes, false);
3264 assert!(matches!(dec, Decoder::StringToBytes(_, _)));
3265 for s in ["", "abc", "data"] {
3266 let data = encode_avro_bytes(s.as_bytes());
3267 let mut cur = AvroCursor::new(&data);
3268 dec.decode(&mut cur).unwrap();
3269 }
3270 let arr = dec.flush(None).unwrap();
3271 let a = arr.as_any().downcast_ref::<BinaryArray>().unwrap();
3272 assert_eq!(a.value(0), b"");
3273 assert_eq!(a.value(1), b"abc");
3274 assert_eq!(a.value(2), b"data");
3275 }
3276
3277 #[test]
3278 fn test_schema_resolution_no_promotion_passthrough_int() {
3279 let ws = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3280 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3281 let writer_record = Schema::Complex(ComplexType::Record(Record {
3283 name: "Root",
3284 namespace: None,
3285 doc: None,
3286 aliases: vec![],
3287 fields: vec![Field {
3288 name: "v",
3289 r#type: ws,
3290 default: None,
3291 doc: None,
3292 aliases: vec![],
3293 }],
3294 attributes: Attributes::default(),
3295 }));
3296 let reader_record = Schema::Complex(ComplexType::Record(Record {
3297 name: "Root",
3298 namespace: None,
3299 doc: None,
3300 aliases: vec![],
3301 fields: vec![Field {
3302 name: "v",
3303 r#type: rs,
3304 default: None,
3305 doc: None,
3306 aliases: vec![],
3307 }],
3308 attributes: Attributes::default(),
3309 }));
3310 let field = AvroFieldBuilder::new(&writer_record)
3311 .with_reader_schema(&reader_record)
3312 .with_utf8view(false)
3313 .with_strict_mode(false)
3314 .build()
3315 .unwrap();
3316 let dt = match field.data_type().codec() {
3318 Codec::Struct(fields) => fields[0].data_type().clone(),
3319 other => panic!("expected wrapper struct, got {other:?}"),
3320 };
3321 let mut dec = Decoder::try_new(&dt).unwrap();
3322 assert!(matches!(dec, Decoder::Int32(_)));
3323 for v in [7, -9] {
3324 let data = encode_avro_int(v);
3325 let mut cur = AvroCursor::new(&data);
3326 dec.decode(&mut cur).unwrap();
3327 }
3328 let arr = dec.flush(None).unwrap();
3329 let a = arr.as_any().downcast_ref::<Int32Array>().unwrap();
3330 assert_eq!(a.value(0), 7);
3331 assert_eq!(a.value(1), -9);
3332 }
3333
3334 #[test]
3335 fn test_schema_resolution_illegal_promotion_int_to_boolean_errors() {
3336 let ws = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3337 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Boolean));
3338 let writer_record = Schema::Complex(ComplexType::Record(Record {
3339 name: "Root",
3340 namespace: None,
3341 doc: None,
3342 aliases: vec![],
3343 fields: vec![Field {
3344 name: "v",
3345 r#type: ws,
3346 default: None,
3347 doc: None,
3348 aliases: vec![],
3349 }],
3350 attributes: Attributes::default(),
3351 }));
3352 let reader_record = Schema::Complex(ComplexType::Record(Record {
3353 name: "Root",
3354 namespace: None,
3355 doc: None,
3356 aliases: vec![],
3357 fields: vec![Field {
3358 name: "v",
3359 r#type: rs,
3360 default: None,
3361 doc: None,
3362 aliases: vec![],
3363 }],
3364 attributes: Attributes::default(),
3365 }));
3366 let res = AvroFieldBuilder::new(&writer_record)
3367 .with_reader_schema(&reader_record)
3368 .with_utf8view(false)
3369 .with_strict_mode(false)
3370 .build();
3371 assert!(res.is_err(), "expected error for illegal promotion");
3372 }
3373
3374 #[test]
3375 fn test_map_decoding_one_entry() {
3376 let value_type = avro_from_codec(Codec::Utf8);
3377 let map_type = avro_from_codec(Codec::Map(Arc::new(value_type)));
3378 let mut decoder = Decoder::try_new(&map_type).unwrap();
3379 let mut data = Vec::new();
3381 data.extend_from_slice(&encode_avro_long(1));
3382 data.extend_from_slice(&encode_avro_bytes(b"hello")); data.extend_from_slice(&encode_avro_bytes(b"world")); data.extend_from_slice(&encode_avro_long(0));
3385 let mut cursor = AvroCursor::new(&data);
3386 decoder.decode(&mut cursor).unwrap();
3387 let array = decoder.flush(None).unwrap();
3388 let map_arr = array.as_any().downcast_ref::<MapArray>().unwrap();
3389 assert_eq!(map_arr.len(), 1); assert_eq!(map_arr.value_length(0), 1);
3391 let entries = map_arr.value(0);
3392 let struct_entries = entries.as_any().downcast_ref::<StructArray>().unwrap();
3393 assert_eq!(struct_entries.len(), 1);
3394 let key_arr = struct_entries
3395 .column_by_name("key")
3396 .unwrap()
3397 .as_any()
3398 .downcast_ref::<StringArray>()
3399 .unwrap();
3400 let val_arr = struct_entries
3401 .column_by_name("value")
3402 .unwrap()
3403 .as_any()
3404 .downcast_ref::<StringArray>()
3405 .unwrap();
3406 assert_eq!(key_arr.value(0), "hello");
3407 assert_eq!(val_arr.value(0), "world");
3408 }
3409
3410 #[test]
3411 fn test_map_decoding_empty() {
3412 let value_type = avro_from_codec(Codec::Utf8);
3413 let map_type = avro_from_codec(Codec::Map(Arc::new(value_type)));
3414 let mut decoder = Decoder::try_new(&map_type).unwrap();
3415 let data = encode_avro_long(0);
3416 decoder.decode(&mut AvroCursor::new(&data)).unwrap();
3417 let array = decoder.flush(None).unwrap();
3418 let map_arr = array.as_any().downcast_ref::<MapArray>().unwrap();
3419 assert_eq!(map_arr.len(), 1);
3420 assert_eq!(map_arr.value_length(0), 0);
3421 }
3422
3423 #[test]
3424 fn test_fixed_decoding() {
3425 let avro_type = avro_from_codec(Codec::Fixed(3));
3426 let mut decoder = Decoder::try_new(&avro_type).expect("Failed to create decoder");
3427
3428 let data1 = [1u8, 2, 3];
3429 let mut cursor1 = AvroCursor::new(&data1);
3430 decoder
3431 .decode(&mut cursor1)
3432 .expect("Failed to decode data1");
3433 assert_eq!(cursor1.position(), 3, "Cursor should advance by fixed size");
3434 let data2 = [4u8, 5, 6];
3435 let mut cursor2 = AvroCursor::new(&data2);
3436 decoder
3437 .decode(&mut cursor2)
3438 .expect("Failed to decode data2");
3439 assert_eq!(cursor2.position(), 3, "Cursor should advance by fixed size");
3440 let array = decoder.flush(None).expect("Failed to flush decoder");
3441 assert_eq!(array.len(), 2, "Array should contain two items");
3442 let fixed_size_binary_array = array
3443 .as_any()
3444 .downcast_ref::<FixedSizeBinaryArray>()
3445 .expect("Failed to downcast to FixedSizeBinaryArray");
3446 assert_eq!(
3447 fixed_size_binary_array.value_length(),
3448 3,
3449 "Fixed size of binary values should be 3"
3450 );
3451 assert_eq!(
3452 fixed_size_binary_array.value(0),
3453 &[1, 2, 3],
3454 "First item mismatch"
3455 );
3456 assert_eq!(
3457 fixed_size_binary_array.value(1),
3458 &[4, 5, 6],
3459 "Second item mismatch"
3460 );
3461 }
3462
3463 #[test]
3464 fn test_fixed_decoding_empty() {
3465 let avro_type = avro_from_codec(Codec::Fixed(5));
3466 let mut decoder = Decoder::try_new(&avro_type).expect("Failed to create decoder");
3467
3468 let array = decoder
3469 .flush(None)
3470 .expect("Failed to flush decoder for empty input");
3471
3472 assert_eq!(array.len(), 0, "Array should be empty");
3473 let fixed_size_binary_array = array
3474 .as_any()
3475 .downcast_ref::<FixedSizeBinaryArray>()
3476 .expect("Failed to downcast to FixedSizeBinaryArray for empty array");
3477
3478 assert_eq!(
3479 fixed_size_binary_array.value_length(),
3480 5,
3481 "Fixed size of binary values should be 5 as per type"
3482 );
3483 }
3484
3485 #[test]
3486 fn test_uuid_decoding() {
3487 let avro_type = avro_from_codec(Codec::Uuid);
3488 let mut decoder = Decoder::try_new(&avro_type).expect("Failed to create decoder");
3489 let uuid_str = "f81d4fae-7dec-11d0-a765-00a0c91e6bf6";
3490 let data = encode_avro_bytes(uuid_str.as_bytes());
3491 let mut cursor = AvroCursor::new(&data);
3492 decoder.decode(&mut cursor).expect("Failed to decode data");
3493 assert_eq!(
3494 cursor.position(),
3495 data.len(),
3496 "Cursor should advance by varint size + data size"
3497 );
3498 let array = decoder.flush(None).expect("Failed to flush decoder");
3499 let fixed_size_binary_array = array
3500 .as_any()
3501 .downcast_ref::<FixedSizeBinaryArray>()
3502 .expect("Array should be a FixedSizeBinaryArray");
3503 assert_eq!(fixed_size_binary_array.len(), 1);
3504 assert_eq!(fixed_size_binary_array.value_length(), 16);
3505 let expected_bytes = [
3506 0xf8, 0x1d, 0x4f, 0xae, 0x7d, 0xec, 0x11, 0xd0, 0xa7, 0x65, 0x00, 0xa0, 0xc9, 0x1e,
3507 0x6b, 0xf6,
3508 ];
3509 assert_eq!(fixed_size_binary_array.value(0), &expected_bytes);
3510 }
3511
3512 #[test]
3513 fn test_array_decoding() {
3514 let item_dt = avro_from_codec(Codec::Int32);
3515 let list_dt = avro_from_codec(Codec::List(Arc::new(item_dt)));
3516 let mut decoder = Decoder::try_new(&list_dt).unwrap();
3517 let mut row1 = Vec::new();
3518 row1.extend_from_slice(&encode_avro_long(2));
3519 row1.extend_from_slice(&encode_avro_int(10));
3520 row1.extend_from_slice(&encode_avro_int(20));
3521 row1.extend_from_slice(&encode_avro_long(0));
3522 let row2 = encode_avro_long(0);
3523 let mut cursor = AvroCursor::new(&row1);
3524 decoder.decode(&mut cursor).unwrap();
3525 let mut cursor2 = AvroCursor::new(&row2);
3526 decoder.decode(&mut cursor2).unwrap();
3527 let array = decoder.flush(None).unwrap();
3528 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3529 assert_eq!(list_arr.len(), 2);
3530 let offsets = list_arr.value_offsets();
3531 assert_eq!(offsets, &[0, 2, 2]);
3532 let values = list_arr.values();
3533 let int_arr = values.as_primitive::<Int32Type>();
3534 assert_eq!(int_arr.len(), 2);
3535 assert_eq!(int_arr.value(0), 10);
3536 assert_eq!(int_arr.value(1), 20);
3537 }
3538
3539 #[test]
3540 fn test_array_decoding_with_negative_block_count() {
3541 let item_dt = avro_from_codec(Codec::Int32);
3542 let list_dt = avro_from_codec(Codec::List(Arc::new(item_dt)));
3543 let mut decoder = Decoder::try_new(&list_dt).unwrap();
3544 let mut data = encode_avro_long(-3);
3545 data.extend_from_slice(&encode_avro_long(12));
3546 data.extend_from_slice(&encode_avro_int(1));
3547 data.extend_from_slice(&encode_avro_int(2));
3548 data.extend_from_slice(&encode_avro_int(3));
3549 data.extend_from_slice(&encode_avro_long(0));
3550 let mut cursor = AvroCursor::new(&data);
3551 decoder.decode(&mut cursor).unwrap();
3552 let array = decoder.flush(None).unwrap();
3553 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3554 assert_eq!(list_arr.len(), 1);
3555 assert_eq!(list_arr.value_length(0), 3);
3556 let values = list_arr.values().as_primitive::<Int32Type>();
3557 assert_eq!(values.len(), 3);
3558 assert_eq!(values.value(0), 1);
3559 assert_eq!(values.value(1), 2);
3560 assert_eq!(values.value(2), 3);
3561 }
3562
3563 fn encode_avro_long_extreme(value: i64) -> Vec<u8> {
3566 let mut n = ((value << 1) ^ (value >> 63)) as u64;
3567 let mut out = Vec::new();
3568 while n >= 0x80 {
3569 out.push((n as u8) | 0x80);
3570 n >>= 7;
3571 }
3572 out.push(n as u8);
3573 out
3574 }
3575
3576 fn array_of_null_decoder() -> Decoder {
3579 let list_dt = avro_from_codec(Codec::List(Arc::new(avro_from_codec(Codec::Null))));
3580 Decoder::try_new(&list_dt).unwrap()
3581 }
3582
3583 #[test]
3584 fn test_array_of_null_decodes() {
3585 let mut decoder = array_of_null_decoder();
3586 let mut data = encode_avro_long(3); data.extend_from_slice(&encode_avro_long(0)); decoder.decode(&mut AvroCursor::new(&data)).unwrap();
3589 }
3590
3591 #[test]
3592 fn test_array_block_count_i64_max_errors() {
3593 let mut decoder = array_of_null_decoder();
3595 let mut data = encode_avro_long_extreme(i64::MAX); data.extend_from_slice(&encode_avro_long(0)); let err = decoder.decode(&mut AvroCursor::new(&data)).unwrap_err();
3598 assert!(
3599 err.to_string().contains("Capacity overflow"),
3600 "unexpected error: {err}",
3601 );
3602 }
3603
3604 #[test]
3605 fn test_array_block_count_i64_min_errors() {
3606 let mut decoder = array_of_null_decoder();
3608 let mut data = encode_avro_long_extreme(i64::MIN); data.extend_from_slice(&encode_avro_long(0)); let err = decoder.decode(&mut AvroCursor::new(&data)).unwrap_err();
3611 assert!(
3612 err.to_string().contains("Capacity overflow"),
3613 "unexpected error: {err}",
3614 );
3615 }
3616
3617 #[test]
3618 fn test_nested_array_decoding() {
3619 let inner_ty = avro_from_codec(Codec::List(Arc::new(avro_from_codec(Codec::Int32))));
3620 let nested_ty = avro_from_codec(Codec::List(Arc::new(inner_ty.clone())));
3621 let mut decoder = Decoder::try_new(&nested_ty).unwrap();
3622 let mut buf = Vec::new();
3623 buf.extend(encode_avro_long(1));
3624 buf.extend(encode_avro_long(2));
3625 buf.extend(encode_avro_int(5));
3626 buf.extend(encode_avro_int(6));
3627 buf.extend(encode_avro_long(0));
3628 buf.extend(encode_avro_long(0));
3629 let mut cursor = AvroCursor::new(&buf);
3630 decoder.decode(&mut cursor).unwrap();
3631 let arr = decoder.flush(None).unwrap();
3632 let outer = arr.as_any().downcast_ref::<ListArray>().unwrap();
3633 assert_eq!(outer.len(), 1);
3634 assert_eq!(outer.value_length(0), 1);
3635 let inner = outer.values().as_any().downcast_ref::<ListArray>().unwrap();
3636 assert_eq!(inner.len(), 1);
3637 assert_eq!(inner.value_length(0), 2);
3638 let values = inner
3639 .values()
3640 .as_any()
3641 .downcast_ref::<Int32Array>()
3642 .unwrap();
3643 assert_eq!(values.values(), &[5, 6]);
3644 }
3645
3646 #[test]
3647 fn test_array_decoding_empty_array() {
3648 let value_type = avro_from_codec(Codec::Utf8);
3649 let map_type = avro_from_codec(Codec::List(Arc::new(value_type)));
3650 let mut decoder = Decoder::try_new(&map_type).unwrap();
3651 let data = encode_avro_long(0);
3652 decoder.decode(&mut AvroCursor::new(&data)).unwrap();
3653 let array = decoder.flush(None).unwrap();
3654 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3655 assert_eq!(list_arr.len(), 1);
3656 assert_eq!(list_arr.value_length(0), 0);
3657 }
3658
3659 #[test]
3660 fn test_array_decoding_writer_nonunion_items_reader_nullable_items() {
3661 use crate::schema::Array;
3662 let writer_schema = Schema::Complex(ComplexType::Array(Array {
3663 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
3664 attributes: Attributes::default(),
3665 }));
3666 let reader_schema = Schema::Complex(ComplexType::Array(Array {
3667 items: Box::new(Schema::Union(vec![
3668 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
3669 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3670 ])),
3671 attributes: Attributes::default(),
3672 }));
3673 let dt = resolved_root_datatype(writer_schema, reader_schema, false, false);
3674 if let Codec::List(inner) = dt.codec() {
3675 assert_eq!(
3676 inner.nullability(),
3677 Some(Nullability::NullFirst),
3678 "items should be nullable"
3679 );
3680 } else {
3681 panic!("expected List codec");
3682 }
3683 let mut decoder = Decoder::try_new(&dt).unwrap();
3684 let mut data = encode_avro_long(2);
3685 data.extend(encode_avro_int(10));
3686 data.extend(encode_avro_int(20));
3687 data.extend(encode_avro_long(0));
3688 let mut cursor = AvroCursor::new(&data);
3689 decoder.decode(&mut cursor).unwrap();
3690 assert_eq!(
3691 cursor.position(),
3692 data.len(),
3693 "all bytes should be consumed"
3694 );
3695 let array = decoder.flush(None).unwrap();
3696 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3697 assert_eq!(list_arr.len(), 1, "one list/row");
3698 assert_eq!(list_arr.value_length(0), 2, "two items in the list");
3699 let values = list_arr.values().as_primitive::<Int32Type>();
3700 assert_eq!(values.len(), 2);
3701 assert_eq!(values.value(0), 10);
3702 assert_eq!(values.value(1), 20);
3703 assert!(!values.is_null(0));
3704 assert!(!values.is_null(1));
3705 }
3706
3707 #[test]
3708 fn test_decimal_decoding_fixed256() {
3709 let dt = avro_from_codec(Codec::Decimal(50, Some(2), Some(32)));
3710 let mut decoder = Decoder::try_new(&dt).unwrap();
3711 let row1 = [
3712 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3713 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3714 0x00, 0x00, 0x30, 0x39,
3715 ];
3716 let row2 = [
3717 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3718 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3719 0xFF, 0xFF, 0xFF, 0x85,
3720 ];
3721 let mut data = Vec::new();
3722 data.extend_from_slice(&row1);
3723 data.extend_from_slice(&row2);
3724 let mut cursor = AvroCursor::new(&data);
3725 decoder.decode(&mut cursor).unwrap();
3726 decoder.decode(&mut cursor).unwrap();
3727 let arr = decoder.flush(None).unwrap();
3728 let dec = arr.as_any().downcast_ref::<Decimal256Array>().unwrap();
3729 assert_eq!(dec.len(), 2);
3730 assert_eq!(dec.value_as_string(0), "123.45");
3731 assert_eq!(dec.value_as_string(1), "-1.23");
3732 }
3733
3734 #[test]
3735 fn test_decimal_decoding_fixed128() {
3736 let dt = avro_from_codec(Codec::Decimal(28, Some(2), Some(16)));
3737 let mut decoder = Decoder::try_new(&dt).unwrap();
3738 let row1 = [
3739 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3740 0x30, 0x39,
3741 ];
3742 let row2 = [
3743 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3744 0xFF, 0x85,
3745 ];
3746 let mut data = Vec::new();
3747 data.extend_from_slice(&row1);
3748 data.extend_from_slice(&row2);
3749 let mut cursor = AvroCursor::new(&data);
3750 decoder.decode(&mut cursor).unwrap();
3751 decoder.decode(&mut cursor).unwrap();
3752 let arr = decoder.flush(None).unwrap();
3753 let dec = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3754 assert_eq!(dec.len(), 2);
3755 assert_eq!(dec.value_as_string(0), "123.45");
3756 assert_eq!(dec.value_as_string(1), "-1.23");
3757 }
3758
3759 #[test]
3760 fn test_decimal_decoding_fixed32_from_32byte_fixed_storage() {
3761 let dt = avro_from_codec(Codec::Decimal(5, Some(2), Some(32)));
3762 let mut decoder = Decoder::try_new(&dt).unwrap();
3763 let row1 = [
3764 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3765 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3766 0x00, 0x00, 0x30, 0x39,
3767 ];
3768 let row2 = [
3769 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3770 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3771 0xFF, 0xFF, 0xFF, 0x85,
3772 ];
3773 let mut data = Vec::new();
3774 data.extend_from_slice(&row1);
3775 data.extend_from_slice(&row2);
3776 let mut cursor = AvroCursor::new(&data);
3777 decoder.decode(&mut cursor).unwrap();
3778 decoder.decode(&mut cursor).unwrap();
3779 let arr = decoder.flush(None).unwrap();
3780 #[cfg(feature = "small_decimals")]
3781 {
3782 let dec = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3783 assert_eq!(dec.len(), 2);
3784 assert_eq!(dec.value_as_string(0), "123.45");
3785 assert_eq!(dec.value_as_string(1), "-1.23");
3786 }
3787 #[cfg(not(feature = "small_decimals"))]
3788 {
3789 let dec = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3790 assert_eq!(dec.len(), 2);
3791 assert_eq!(dec.value_as_string(0), "123.45");
3792 assert_eq!(dec.value_as_string(1), "-1.23");
3793 }
3794 }
3795
3796 #[test]
3797 fn test_decimal_decoding_fixed32_from_16byte_fixed_storage() {
3798 let dt = avro_from_codec(Codec::Decimal(5, Some(2), Some(16)));
3799 let mut decoder = Decoder::try_new(&dt).unwrap();
3800 let row1 = [
3801 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3802 0x30, 0x39,
3803 ];
3804 let row2 = [
3805 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3806 0xFF, 0x85,
3807 ];
3808 let mut data = Vec::new();
3809 data.extend_from_slice(&row1);
3810 data.extend_from_slice(&row2);
3811 let mut cursor = AvroCursor::new(&data);
3812 decoder.decode(&mut cursor).unwrap();
3813 decoder.decode(&mut cursor).unwrap();
3814
3815 let arr = decoder.flush(None).unwrap();
3816 #[cfg(feature = "small_decimals")]
3817 {
3818 let dec = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3819 assert_eq!(dec.len(), 2);
3820 assert_eq!(dec.value_as_string(0), "123.45");
3821 assert_eq!(dec.value_as_string(1), "-1.23");
3822 }
3823 #[cfg(not(feature = "small_decimals"))]
3824 {
3825 let dec = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3826 assert_eq!(dec.len(), 2);
3827 assert_eq!(dec.value_as_string(0), "123.45");
3828 assert_eq!(dec.value_as_string(1), "-1.23");
3829 }
3830 }
3831
3832 #[test]
3833 fn test_decimal_decoding_bytes_with_nulls() {
3834 let dt = avro_from_codec(Codec::Decimal(4, Some(1), None));
3835 let inner = Decoder::try_new(&dt).unwrap();
3836 let mut decoder = Decoder::Nullable(NullableDecoder::new(
3837 NullablePlan::ReadTag {
3838 nullability: Nullability::NullSecond,
3839 resolution: ResolutionPlan::Promotion(Promotion::Direct),
3840 },
3841 inner,
3842 ));
3843 let mut data = Vec::new();
3844 data.extend_from_slice(&encode_avro_int(0));
3845 data.extend_from_slice(&encode_avro_bytes(&[0x04, 0xD2]));
3846 data.extend_from_slice(&encode_avro_int(1));
3847 data.extend_from_slice(&encode_avro_int(0));
3848 data.extend_from_slice(&encode_avro_bytes(&[0xFB, 0x2E]));
3849 let mut cursor = AvroCursor::new(&data);
3850 decoder.decode(&mut cursor).unwrap();
3851 decoder.decode(&mut cursor).unwrap();
3852 decoder.decode(&mut cursor).unwrap();
3853 let arr = decoder.flush(None).unwrap();
3854 #[cfg(feature = "small_decimals")]
3855 {
3856 let dec_arr = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3857 assert_eq!(dec_arr.len(), 3);
3858 assert!(dec_arr.is_valid(0));
3859 assert!(!dec_arr.is_valid(1));
3860 assert!(dec_arr.is_valid(2));
3861 assert_eq!(dec_arr.value_as_string(0), "123.4");
3862 assert_eq!(dec_arr.value_as_string(2), "-123.4");
3863 }
3864 #[cfg(not(feature = "small_decimals"))]
3865 {
3866 let dec_arr = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3867 assert_eq!(dec_arr.len(), 3);
3868 assert!(dec_arr.is_valid(0));
3869 assert!(!dec_arr.is_valid(1));
3870 assert!(dec_arr.is_valid(2));
3871 assert_eq!(dec_arr.value_as_string(0), "123.4");
3872 assert_eq!(dec_arr.value_as_string(2), "-123.4");
3873 }
3874 }
3875
3876 #[test]
3877 fn test_decimal_decoding_bytes_with_nulls_fixed_size_narrow_result() {
3878 let dt = avro_from_codec(Codec::Decimal(6, Some(2), Some(16)));
3879 let inner = Decoder::try_new(&dt).unwrap();
3880 let mut decoder = Decoder::Nullable(NullableDecoder::new(
3881 NullablePlan::ReadTag {
3882 nullability: Nullability::NullSecond,
3883 resolution: ResolutionPlan::Promotion(Promotion::Direct),
3884 },
3885 inner,
3886 ));
3887 let row1 = [
3888 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
3889 0xE2, 0x40,
3890 ];
3891 let row3 = [
3892 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFE,
3893 0x1D, 0xC0,
3894 ];
3895 let mut data = Vec::new();
3896 data.extend_from_slice(&encode_avro_int(0));
3897 data.extend_from_slice(&row1);
3898 data.extend_from_slice(&encode_avro_int(1));
3899 data.extend_from_slice(&encode_avro_int(0));
3900 data.extend_from_slice(&row3);
3901 let mut cursor = AvroCursor::new(&data);
3902 decoder.decode(&mut cursor).unwrap();
3903 decoder.decode(&mut cursor).unwrap();
3904 decoder.decode(&mut cursor).unwrap();
3905 let arr = decoder.flush(None).unwrap();
3906 #[cfg(feature = "small_decimals")]
3907 {
3908 let dec_arr = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3909 assert_eq!(dec_arr.len(), 3);
3910 assert!(dec_arr.is_valid(0));
3911 assert!(!dec_arr.is_valid(1));
3912 assert!(dec_arr.is_valid(2));
3913 assert_eq!(dec_arr.value_as_string(0), "1234.56");
3914 assert_eq!(dec_arr.value_as_string(2), "-1234.56");
3915 }
3916 #[cfg(not(feature = "small_decimals"))]
3917 {
3918 let dec_arr = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3919 assert_eq!(dec_arr.len(), 3);
3920 assert!(dec_arr.is_valid(0));
3921 assert!(!dec_arr.is_valid(1));
3922 assert!(dec_arr.is_valid(2));
3923 assert_eq!(dec_arr.value_as_string(0), "1234.56");
3924 assert_eq!(dec_arr.value_as_string(2), "-1234.56");
3925 }
3926 }
3927
3928 #[test]
3929 fn test_enum_decoding() {
3930 let symbols: Arc<[String]> = vec!["A", "B", "C"].into_iter().map(String::from).collect();
3931 let avro_type = avro_from_codec(Codec::Enum(symbols.clone()));
3932 let mut decoder = Decoder::try_new(&avro_type).unwrap();
3933 let mut data = Vec::new();
3934 data.extend_from_slice(&encode_avro_int(2));
3935 data.extend_from_slice(&encode_avro_int(0));
3936 data.extend_from_slice(&encode_avro_int(1));
3937 let mut cursor = AvroCursor::new(&data);
3938 decoder.decode(&mut cursor).unwrap();
3939 decoder.decode(&mut cursor).unwrap();
3940 decoder.decode(&mut cursor).unwrap();
3941 let array = decoder.flush(None).unwrap();
3942 let dict_array = array
3943 .as_any()
3944 .downcast_ref::<DictionaryArray<Int32Type>>()
3945 .unwrap();
3946 assert_eq!(dict_array.len(), 3);
3947 let values = dict_array
3948 .values()
3949 .as_any()
3950 .downcast_ref::<StringArray>()
3951 .unwrap();
3952 assert_eq!(values.value(0), "A");
3953 assert_eq!(values.value(1), "B");
3954 assert_eq!(values.value(2), "C");
3955 assert_eq!(dict_array.keys().values(), &[2, 0, 1]);
3956 }
3957
3958 #[test]
3959 fn test_enum_decoding_with_nulls() {
3960 let symbols: Arc<[String]> = vec!["X", "Y"].into_iter().map(String::from).collect();
3961 let enum_codec = Codec::Enum(symbols.clone());
3962 let avro_type =
3963 AvroDataType::new(enum_codec, Default::default(), Some(Nullability::NullFirst));
3964 let mut decoder = Decoder::try_new(&avro_type).unwrap();
3965 let mut data = Vec::new();
3966 data.extend_from_slice(&encode_avro_long(1));
3967 data.extend_from_slice(&encode_avro_int(1));
3968 data.extend_from_slice(&encode_avro_long(0));
3969 data.extend_from_slice(&encode_avro_long(1));
3970 data.extend_from_slice(&encode_avro_int(0));
3971 let mut cursor = AvroCursor::new(&data);
3972 decoder.decode(&mut cursor).unwrap();
3973 decoder.decode(&mut cursor).unwrap();
3974 decoder.decode(&mut cursor).unwrap();
3975 let array = decoder.flush(None).unwrap();
3976 let dict_array = array
3977 .as_any()
3978 .downcast_ref::<DictionaryArray<Int32Type>>()
3979 .unwrap();
3980 assert_eq!(dict_array.len(), 3);
3981 assert!(dict_array.is_valid(0));
3982 assert!(dict_array.is_null(1));
3983 assert!(dict_array.is_valid(2));
3984 let expected_keys = Int32Array::from(vec![Some(1), None, Some(0)]);
3985 assert_eq!(dict_array.keys(), &expected_keys);
3986 let values = dict_array
3987 .values()
3988 .as_any()
3989 .downcast_ref::<StringArray>()
3990 .unwrap();
3991 assert_eq!(values.value(0), "X");
3992 assert_eq!(values.value(1), "Y");
3993 }
3994
3995 #[test]
3996 fn test_duration_decoding_with_nulls() {
3997 let duration_codec = Codec::Interval;
3998 let avro_type = AvroDataType::new(
3999 duration_codec,
4000 Default::default(),
4001 Some(Nullability::NullFirst),
4002 );
4003 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4004 let mut data = Vec::new();
4005 data.extend_from_slice(&encode_avro_long(1)); let mut duration1 = Vec::new();
4008 duration1.extend_from_slice(&1u32.to_le_bytes());
4009 duration1.extend_from_slice(&2u32.to_le_bytes());
4010 duration1.extend_from_slice(&3u32.to_le_bytes());
4011 data.extend_from_slice(&duration1);
4012 data.extend_from_slice(&encode_avro_long(0)); data.extend_from_slice(&encode_avro_long(1)); let mut duration2 = Vec::new();
4016 duration2.extend_from_slice(&4u32.to_le_bytes());
4017 duration2.extend_from_slice(&5u32.to_le_bytes());
4018 duration2.extend_from_slice(&6u32.to_le_bytes());
4019 data.extend_from_slice(&duration2);
4020 let mut cursor = AvroCursor::new(&data);
4021 decoder.decode(&mut cursor).unwrap();
4022 decoder.decode(&mut cursor).unwrap();
4023 decoder.decode(&mut cursor).unwrap();
4024 let array = decoder.flush(None).unwrap();
4025 let interval_array = array
4026 .as_any()
4027 .downcast_ref::<IntervalMonthDayNanoArray>()
4028 .unwrap();
4029 assert_eq!(interval_array.len(), 3);
4030 assert!(interval_array.is_valid(0));
4031 assert!(interval_array.is_null(1));
4032 assert!(interval_array.is_valid(2));
4033 let expected = IntervalMonthDayNanoArray::from(vec![
4034 Some(IntervalMonthDayNano {
4035 months: 1,
4036 days: 2,
4037 nanoseconds: 3_000_000,
4038 }),
4039 None,
4040 Some(IntervalMonthDayNano {
4041 months: 4,
4042 days: 5,
4043 nanoseconds: 6_000_000,
4044 }),
4045 ]);
4046 assert_eq!(interval_array, &expected);
4047 }
4048
4049 #[cfg(feature = "avro_custom_types")]
4050 #[test]
4051 fn test_interval_month_day_nano_custom_decoding_with_nulls() {
4052 let avro_type = AvroDataType::new(
4053 Codec::IntervalMonthDayNano,
4054 Default::default(),
4055 Some(Nullability::NullFirst),
4056 );
4057 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4058 let mut data = Vec::new();
4059 data.extend_from_slice(&encode_avro_long(1));
4061 data.extend_from_slice(&1i32.to_le_bytes());
4062 data.extend_from_slice(&(-2i32).to_le_bytes());
4063 data.extend_from_slice(&3i64.to_le_bytes());
4064 data.extend_from_slice(&encode_avro_long(0));
4066 data.extend_from_slice(&encode_avro_long(1));
4068 data.extend_from_slice(&(-4i32).to_le_bytes());
4069 data.extend_from_slice(&5i32.to_le_bytes());
4070 data.extend_from_slice(&(-6i64).to_le_bytes());
4071 let mut cursor = AvroCursor::new(&data);
4072 decoder.decode(&mut cursor).unwrap();
4073 decoder.decode(&mut cursor).unwrap();
4074 decoder.decode(&mut cursor).unwrap();
4075 let array = decoder.flush(None).unwrap();
4076 let interval_array = array
4077 .as_any()
4078 .downcast_ref::<IntervalMonthDayNanoArray>()
4079 .unwrap();
4080 assert_eq!(interval_array.len(), 3);
4081 let expected = IntervalMonthDayNanoArray::from(vec![
4082 Some(IntervalMonthDayNano::new(1, -2, 3)),
4083 None,
4084 Some(IntervalMonthDayNano::new(-4, 5, -6)),
4085 ]);
4086 assert_eq!(interval_array, &expected);
4087 }
4088
4089 #[test]
4090 fn test_duration_decoding_empty() {
4091 let duration_codec = Codec::Interval;
4092 let avro_type = AvroDataType::new(duration_codec, Default::default(), None);
4093 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4094 let array = decoder.flush(None).unwrap();
4095 assert_eq!(array.len(), 0);
4096 }
4097
4098 #[test]
4099 #[cfg(feature = "avro_custom_types")]
4100 fn test_duration_seconds_decoding() {
4101 let avro_type = AvroDataType::new(Codec::DurationSeconds, Default::default(), None);
4102 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4103 let mut data = Vec::new();
4104 data.extend_from_slice(&encode_avro_long(0));
4106 data.extend_from_slice(&encode_avro_long(-1));
4107 data.extend_from_slice(&encode_avro_long(2));
4108 let mut cursor = AvroCursor::new(&data);
4109 decoder.decode(&mut cursor).unwrap();
4110 decoder.decode(&mut cursor).unwrap();
4111 decoder.decode(&mut cursor).unwrap();
4112 let array = decoder.flush(None).unwrap();
4113 let dur = array
4114 .as_any()
4115 .downcast_ref::<DurationSecondArray>()
4116 .unwrap();
4117 assert_eq!(dur.values(), &[0, -1, 2]);
4118 }
4119
4120 #[test]
4121 #[cfg(feature = "avro_custom_types")]
4122 fn test_duration_milliseconds_decoding() {
4123 let avro_type = AvroDataType::new(Codec::DurationMillis, Default::default(), None);
4124 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4125 let mut data = Vec::new();
4126 for v in [1i64, 0, -2] {
4127 data.extend_from_slice(&encode_avro_long(v));
4128 }
4129 let mut cursor = AvroCursor::new(&data);
4130 for _ in 0..3 {
4131 decoder.decode(&mut cursor).unwrap();
4132 }
4133 let array = decoder.flush(None).unwrap();
4134 let dur = array
4135 .as_any()
4136 .downcast_ref::<DurationMillisecondArray>()
4137 .unwrap();
4138 assert_eq!(dur.values(), &[1, 0, -2]);
4139 }
4140
4141 #[test]
4142 #[cfg(feature = "avro_custom_types")]
4143 fn test_duration_microseconds_decoding() {
4144 let avro_type = AvroDataType::new(Codec::DurationMicros, Default::default(), None);
4145 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4146 let mut data = Vec::new();
4147 for v in [5i64, -6, 7] {
4148 data.extend_from_slice(&encode_avro_long(v));
4149 }
4150 let mut cursor = AvroCursor::new(&data);
4151 for _ in 0..3 {
4152 decoder.decode(&mut cursor).unwrap();
4153 }
4154 let array = decoder.flush(None).unwrap();
4155 let dur = array
4156 .as_any()
4157 .downcast_ref::<DurationMicrosecondArray>()
4158 .unwrap();
4159 assert_eq!(dur.values(), &[5, -6, 7]);
4160 }
4161
4162 #[test]
4163 #[cfg(feature = "avro_custom_types")]
4164 fn test_duration_nanoseconds_decoding() {
4165 let avro_type = AvroDataType::new(Codec::DurationNanos, Default::default(), None);
4166 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4167 let mut data = Vec::new();
4168 for v in [8i64, 9, -10] {
4169 data.extend_from_slice(&encode_avro_long(v));
4170 }
4171 let mut cursor = AvroCursor::new(&data);
4172 for _ in 0..3 {
4173 decoder.decode(&mut cursor).unwrap();
4174 }
4175 let array = decoder.flush(None).unwrap();
4176 let dur = array
4177 .as_any()
4178 .downcast_ref::<DurationNanosecondArray>()
4179 .unwrap();
4180 assert_eq!(dur.values(), &[8, 9, -10]);
4181 }
4182
4183 #[test]
4184 fn test_nullable_decode_error_bitmap_corruption() {
4185 let avro_type = AvroDataType::new(
4187 Codec::Int32,
4188 Default::default(),
4189 Some(Nullability::NullSecond),
4190 );
4191 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4192
4193 let mut row1 = Vec::new();
4195 row1.extend_from_slice(&encode_avro_int(1));
4196
4197 let mut row2 = Vec::new();
4199 row2.extend_from_slice(&encode_avro_int(0)); let mut row3 = Vec::new();
4203 row3.extend_from_slice(&encode_avro_int(0)); row3.extend_from_slice(&encode_avro_int(42)); decoder.decode(&mut AvroCursor::new(&row1)).unwrap();
4207 assert!(decoder.decode(&mut AvroCursor::new(&row2)).is_err()); decoder.decode(&mut AvroCursor::new(&row3)).unwrap();
4209
4210 let array = decoder.flush(None).unwrap();
4211
4212 assert_eq!(array.len(), 2);
4214 let int_array = array.as_any().downcast_ref::<Int32Array>().unwrap();
4215 assert!(int_array.is_null(0)); assert_eq!(int_array.value(1), 42); }
4218
4219 #[test]
4220 fn test_enum_mapping_reordered_symbols() {
4221 let reader_symbols: Arc<[String]> =
4222 vec!["B".to_string(), "C".to_string(), "A".to_string()].into();
4223 let mapping: Arc<[i32]> = Arc::from(vec![2, 0, 1]);
4224 let default_index: i32 = -1;
4225 let mut dec = Decoder::Enum(
4226 Vec::with_capacity(DEFAULT_CAPACITY),
4227 reader_symbols.clone(),
4228 Some(EnumResolution {
4229 mapping,
4230 default_index,
4231 }),
4232 );
4233 let mut data = Vec::new();
4234 data.extend_from_slice(&encode_avro_int(0));
4235 data.extend_from_slice(&encode_avro_int(1));
4236 data.extend_from_slice(&encode_avro_int(2));
4237 let mut cur = AvroCursor::new(&data);
4238 dec.decode(&mut cur).unwrap();
4239 dec.decode(&mut cur).unwrap();
4240 dec.decode(&mut cur).unwrap();
4241 let arr = dec.flush(None).unwrap();
4242 let dict = arr
4243 .as_any()
4244 .downcast_ref::<DictionaryArray<Int32Type>>()
4245 .unwrap();
4246 let expected_keys = Int32Array::from(vec![2, 0, 1]);
4247 assert_eq!(dict.keys(), &expected_keys);
4248 let values = dict
4249 .values()
4250 .as_any()
4251 .downcast_ref::<StringArray>()
4252 .unwrap();
4253 assert_eq!(values.value(0), "B");
4254 assert_eq!(values.value(1), "C");
4255 assert_eq!(values.value(2), "A");
4256 }
4257
4258 #[test]
4259 fn test_enum_mapping_unknown_symbol_and_out_of_range_fall_back_to_default() {
4260 let reader_symbols: Arc<[String]> = vec!["A".to_string(), "B".to_string()].into();
4261 let default_index: i32 = 1;
4262 let mapping: Arc<[i32]> = Arc::from(vec![0, 1]);
4263 let mut dec = Decoder::Enum(
4264 Vec::with_capacity(DEFAULT_CAPACITY),
4265 reader_symbols.clone(),
4266 Some(EnumResolution {
4267 mapping,
4268 default_index,
4269 }),
4270 );
4271 let mut data = Vec::new();
4272 data.extend_from_slice(&encode_avro_int(0));
4273 data.extend_from_slice(&encode_avro_int(1));
4274 data.extend_from_slice(&encode_avro_int(99));
4275 let mut cur = AvroCursor::new(&data);
4276 dec.decode(&mut cur).unwrap();
4277 dec.decode(&mut cur).unwrap();
4278 dec.decode(&mut cur).unwrap();
4279 let arr = dec.flush(None).unwrap();
4280 let dict = arr
4281 .as_any()
4282 .downcast_ref::<DictionaryArray<Int32Type>>()
4283 .unwrap();
4284 let expected_keys = Int32Array::from(vec![0, 1, 1]);
4285 assert_eq!(dict.keys(), &expected_keys);
4286 let values = dict
4287 .values()
4288 .as_any()
4289 .downcast_ref::<StringArray>()
4290 .unwrap();
4291 assert_eq!(values.value(0), "A");
4292 assert_eq!(values.value(1), "B");
4293 }
4294
4295 #[test]
4296 fn test_enum_mapping_unknown_symbol_without_default_errors() {
4297 let reader_symbols: Arc<[String]> = vec!["A".to_string()].into();
4298 let default_index: i32 = -1; let mapping: Arc<[i32]> = Arc::from(vec![-1]);
4300 let mut dec = Decoder::Enum(
4301 Vec::with_capacity(DEFAULT_CAPACITY),
4302 reader_symbols,
4303 Some(EnumResolution {
4304 mapping,
4305 default_index,
4306 }),
4307 );
4308 let data = encode_avro_int(0);
4309 let mut cur = AvroCursor::new(&data);
4310 let err = dec
4311 .decode(&mut cur)
4312 .expect_err("expected decode error for unresolved enum without default");
4313 let msg = err.to_string();
4314 assert!(
4315 msg.contains("not resolvable") && msg.contains("no default"),
4316 "unexpected error message: {msg}"
4317 );
4318 }
4319
4320 fn make_record_resolved_decoder(
4321 reader_fields: &[(&str, DataType, bool)],
4322 writer_projections: Vec<FieldProjection>,
4323 ) -> Decoder {
4324 let mut field_refs: Vec<FieldRef> = Vec::with_capacity(reader_fields.len());
4325 let mut encodings: Vec<Decoder> = Vec::with_capacity(reader_fields.len());
4326 for (name, dt, nullable) in reader_fields {
4327 field_refs.push(Arc::new(ArrowField::new(*name, dt.clone(), *nullable)));
4328 let enc = match dt {
4329 DataType::Int32 => Decoder::Int32(Vec::new()),
4330 DataType::Int64 => Decoder::Int64(Vec::new()),
4331 DataType::Utf8 => {
4332 Decoder::String(OffsetBufferBuilder::new(DEFAULT_CAPACITY), Vec::new())
4333 }
4334 other => panic!("Unsupported test reader field type: {other:?}"),
4335 };
4336 encodings.push(enc);
4337 }
4338 let fields: Fields = field_refs.into();
4339 Decoder::Record {
4340 fields,
4341 decoders: encodings,
4342 defaults: vec![None; reader_fields.len()],
4343 projector: Some(Projector {
4344 writer_projections,
4345 default_injections: Arc::from(Vec::<(usize, AvroLiteral)>::new()),
4346 }),
4347 row_count: 0,
4348 }
4349 }
4350
4351 #[test]
4352 fn test_skip_writer_trailing_field_int32() {
4353 let mut dec = make_record_resolved_decoder(
4354 &[("id", arrow_schema::DataType::Int32, false)],
4355 vec![
4356 FieldProjection::ToReader(0),
4357 FieldProjection::Skip(super::Skipper::Int32),
4358 ],
4359 );
4360 let mut data = Vec::new();
4361 data.extend_from_slice(&encode_avro_int(7));
4362 data.extend_from_slice(&encode_avro_int(999));
4363 let mut cur = AvroCursor::new(&data);
4364 dec.decode(&mut cur).unwrap();
4365 assert_eq!(cur.position(), data.len());
4366 let arr = dec.flush(None).unwrap();
4367 let struct_arr = arr.as_any().downcast_ref::<StructArray>().unwrap();
4368 assert_eq!(struct_arr.len(), 1);
4369 let id = struct_arr
4370 .column_by_name("id")
4371 .unwrap()
4372 .as_any()
4373 .downcast_ref::<Int32Array>()
4374 .unwrap();
4375 assert_eq!(id.value(0), 7);
4376 }
4377
4378 #[test]
4379 fn test_skip_writer_middle_field_string() {
4380 let mut dec = make_record_resolved_decoder(
4381 &[
4382 ("id", DataType::Int32, false),
4383 ("score", DataType::Int64, false),
4384 ],
4385 vec![
4386 FieldProjection::ToReader(0),
4387 FieldProjection::Skip(Skipper::String),
4388 FieldProjection::ToReader(1),
4389 ],
4390 );
4391 let mut data = Vec::new();
4392 data.extend_from_slice(&encode_avro_int(42));
4393 data.extend_from_slice(&encode_avro_bytes(b"abcdef"));
4394 data.extend_from_slice(&encode_avro_long(1000));
4395 let mut cur = AvroCursor::new(&data);
4396 dec.decode(&mut cur).unwrap();
4397 assert_eq!(cur.position(), data.len());
4398 let arr = dec.flush(None).unwrap();
4399 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4400 let id = s
4401 .column_by_name("id")
4402 .unwrap()
4403 .as_any()
4404 .downcast_ref::<Int32Array>()
4405 .unwrap();
4406 let score = s
4407 .column_by_name("score")
4408 .unwrap()
4409 .as_any()
4410 .downcast_ref::<Int64Array>()
4411 .unwrap();
4412 assert_eq!(id.value(0), 42);
4413 assert_eq!(score.value(0), 1000);
4414 }
4415
4416 #[test]
4417 fn test_skip_writer_array_with_negative_block_count_fast() {
4418 let mut dec = make_record_resolved_decoder(
4419 &[("id", DataType::Int32, false)],
4420 vec![
4421 FieldProjection::Skip(super::Skipper::List(Box::new(Skipper::Int32))),
4422 FieldProjection::ToReader(0),
4423 ],
4424 );
4425 let mut array_payload = Vec::new();
4426 array_payload.extend_from_slice(&encode_avro_int(1));
4427 array_payload.extend_from_slice(&encode_avro_int(2));
4428 array_payload.extend_from_slice(&encode_avro_int(3));
4429 let mut data = Vec::new();
4430 data.extend_from_slice(&encode_avro_long(-3));
4431 data.extend_from_slice(&encode_avro_long(array_payload.len() as i64));
4432 data.extend_from_slice(&array_payload);
4433 data.extend_from_slice(&encode_avro_long(0));
4434 data.extend_from_slice(&encode_avro_int(5));
4435 let mut cur = AvroCursor::new(&data);
4436 dec.decode(&mut cur).unwrap();
4437 assert_eq!(cur.position(), data.len());
4438 let arr = dec.flush(None).unwrap();
4439 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4440 let id = s
4441 .column_by_name("id")
4442 .unwrap()
4443 .as_any()
4444 .downcast_ref::<Int32Array>()
4445 .unwrap();
4446 assert_eq!(id.len(), 1);
4447 assert_eq!(id.value(0), 5);
4448 }
4449
4450 #[test]
4451 fn test_skip_writer_map_with_negative_block_count_fast() {
4452 let mut dec = make_record_resolved_decoder(
4453 &[("id", DataType::Int32, false)],
4454 vec![
4455 FieldProjection::Skip(Skipper::Map(Box::new(Skipper::Int32))),
4456 FieldProjection::ToReader(0),
4457 ],
4458 );
4459 let mut entries = Vec::new();
4460 entries.extend_from_slice(&encode_avro_bytes(b"k1"));
4461 entries.extend_from_slice(&encode_avro_int(10));
4462 entries.extend_from_slice(&encode_avro_bytes(b"k2"));
4463 entries.extend_from_slice(&encode_avro_int(20));
4464 let mut data = Vec::new();
4465 data.extend_from_slice(&encode_avro_long(-2));
4466 data.extend_from_slice(&encode_avro_long(entries.len() as i64));
4467 data.extend_from_slice(&entries);
4468 data.extend_from_slice(&encode_avro_long(0));
4469 data.extend_from_slice(&encode_avro_int(123));
4470 let mut cur = AvroCursor::new(&data);
4471 dec.decode(&mut cur).unwrap();
4472 assert_eq!(cur.position(), data.len());
4473 let arr = dec.flush(None).unwrap();
4474 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4475 let id = s
4476 .column_by_name("id")
4477 .unwrap()
4478 .as_any()
4479 .downcast_ref::<Int32Array>()
4480 .unwrap();
4481 assert_eq!(id.len(), 1);
4482 assert_eq!(id.value(0), 123);
4483 }
4484
4485 #[test]
4486 fn test_skip_writer_nullable_field_union_nullfirst() {
4487 let mut dec = make_record_resolved_decoder(
4488 &[("id", DataType::Int32, false)],
4489 vec![
4490 FieldProjection::Skip(super::Skipper::Nullable(
4491 Nullability::NullFirst,
4492 Box::new(super::Skipper::Int32),
4493 )),
4494 FieldProjection::ToReader(0),
4495 ],
4496 );
4497 let mut row1 = Vec::new();
4498 row1.extend_from_slice(&encode_avro_long(0));
4499 row1.extend_from_slice(&encode_avro_int(5));
4500 let mut row2 = Vec::new();
4501 row2.extend_from_slice(&encode_avro_long(1));
4502 row2.extend_from_slice(&encode_avro_int(123));
4503 row2.extend_from_slice(&encode_avro_int(7));
4504 let mut cur1 = AvroCursor::new(&row1);
4505 let mut cur2 = AvroCursor::new(&row2);
4506 dec.decode(&mut cur1).unwrap();
4507 dec.decode(&mut cur2).unwrap();
4508 assert_eq!(cur1.position(), row1.len());
4509 assert_eq!(cur2.position(), row2.len());
4510 let arr = dec.flush(None).unwrap();
4511 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4512 let id = s
4513 .column_by_name("id")
4514 .unwrap()
4515 .as_any()
4516 .downcast_ref::<Int32Array>()
4517 .unwrap();
4518 assert_eq!(id.len(), 2);
4519 assert_eq!(id.value(0), 5);
4520 assert_eq!(id.value(1), 7);
4521 }
4522
4523 fn make_dense_union_avro(
4524 children: Vec<(Codec, &'_ str, DataType)>,
4525 type_ids: Vec<i8>,
4526 ) -> AvroDataType {
4527 let mut avro_children: Vec<AvroDataType> = Vec::with_capacity(children.len());
4528 let mut fields: Vec<arrow_schema::Field> = Vec::with_capacity(children.len());
4529 for (codec, name, dt) in children {
4530 avro_children.push(AvroDataType::new(codec, Default::default(), None));
4531 fields.push(arrow_schema::Field::new(name, dt, true));
4532 }
4533 let union_fields = UnionFields::try_new(type_ids, fields).unwrap();
4534 let union_codec = Codec::Union(avro_children.into(), union_fields, UnionMode::Dense);
4535 AvroDataType::new(union_codec, Default::default(), None)
4536 }
4537
4538 #[test]
4539 fn test_union_dense_two_children_custom_type_ids() {
4540 let union_dt = make_dense_union_avro(
4541 vec![
4542 (Codec::Int32, "i", DataType::Int32),
4543 (Codec::Utf8, "s", DataType::Utf8),
4544 ],
4545 vec![2, 5],
4546 );
4547 let mut dec = Decoder::try_new(&union_dt).unwrap();
4548 let mut r1 = Vec::new();
4549 r1.extend_from_slice(&encode_avro_long(0));
4550 r1.extend_from_slice(&encode_avro_int(7));
4551 let mut r2 = Vec::new();
4552 r2.extend_from_slice(&encode_avro_long(1));
4553 r2.extend_from_slice(&encode_avro_bytes(b"x"));
4554 let mut r3 = Vec::new();
4555 r3.extend_from_slice(&encode_avro_long(0));
4556 r3.extend_from_slice(&encode_avro_int(-1));
4557 dec.decode(&mut AvroCursor::new(&r1)).unwrap();
4558 dec.decode(&mut AvroCursor::new(&r2)).unwrap();
4559 dec.decode(&mut AvroCursor::new(&r3)).unwrap();
4560 let array = dec.flush(None).unwrap();
4561 let ua = array
4562 .as_any()
4563 .downcast_ref::<UnionArray>()
4564 .expect("expected UnionArray");
4565 assert_eq!(ua.len(), 3);
4566 assert_eq!(ua.type_id(0), 2);
4567 assert_eq!(ua.type_id(1), 5);
4568 assert_eq!(ua.type_id(2), 2);
4569 assert_eq!(ua.value_offset(0), 0);
4570 assert_eq!(ua.value_offset(1), 0);
4571 assert_eq!(ua.value_offset(2), 1);
4572 let int_child = ua
4573 .child(2)
4574 .as_any()
4575 .downcast_ref::<Int32Array>()
4576 .expect("int child");
4577 assert_eq!(int_child.len(), 2);
4578 assert_eq!(int_child.value(0), 7);
4579 assert_eq!(int_child.value(1), -1);
4580 let str_child = ua
4581 .child(5)
4582 .as_any()
4583 .downcast_ref::<StringArray>()
4584 .expect("string child");
4585 assert_eq!(str_child.len(), 1);
4586 assert_eq!(str_child.value(0), "x");
4587 }
4588
4589 #[test]
4590 fn test_union_dense_with_null_and_string_children() {
4591 let union_dt = make_dense_union_avro(
4592 vec![
4593 (Codec::Null, "n", DataType::Null),
4594 (Codec::Utf8, "s", DataType::Utf8),
4595 ],
4596 vec![42, 7],
4597 );
4598 let mut dec = Decoder::try_new(&union_dt).unwrap();
4599 let r1 = encode_avro_long(0);
4600 let mut r2 = Vec::new();
4601 r2.extend_from_slice(&encode_avro_long(1));
4602 r2.extend_from_slice(&encode_avro_bytes(b"abc"));
4603 let r3 = encode_avro_long(0);
4604 dec.decode(&mut AvroCursor::new(&r1)).unwrap();
4605 dec.decode(&mut AvroCursor::new(&r2)).unwrap();
4606 dec.decode(&mut AvroCursor::new(&r3)).unwrap();
4607 let array = dec.flush(None).unwrap();
4608 let ua = array
4609 .as_any()
4610 .downcast_ref::<UnionArray>()
4611 .expect("expected UnionArray");
4612 assert_eq!(ua.len(), 3);
4613 assert_eq!(ua.type_id(0), 42);
4614 assert_eq!(ua.type_id(1), 7);
4615 assert_eq!(ua.type_id(2), 42);
4616 assert_eq!(ua.value_offset(0), 0);
4617 assert_eq!(ua.value_offset(1), 0);
4618 assert_eq!(ua.value_offset(2), 1);
4619 let null_child = ua
4620 .child(42)
4621 .as_any()
4622 .downcast_ref::<NullArray>()
4623 .expect("null child");
4624 assert_eq!(null_child.len(), 2);
4625 let str_child = ua
4626 .child(7)
4627 .as_any()
4628 .downcast_ref::<StringArray>()
4629 .expect("string child");
4630 assert_eq!(str_child.len(), 1);
4631 assert_eq!(str_child.value(0), "abc");
4632 }
4633
4634 #[test]
4635 fn test_union_dense_offsets_reset_after_flush() {
4636 let union_dt = make_dense_union_avro(
4637 vec![
4638 (Codec::Null, "n", DataType::Null),
4639 (Codec::Utf8, "s", DataType::Utf8),
4640 ],
4641 vec![0, 1],
4642 );
4643 let mut dec = Decoder::try_new(&union_dt).unwrap();
4644
4645 for _ in 0..2 {
4646 dec.decode(&mut AvroCursor::new(&encode_avro_long(0)))
4647 .unwrap();
4648 let array = dec.flush(None).unwrap();
4649 let union = array.as_union();
4650 assert_eq!(union.value_offset(0), 0);
4651 }
4652 }
4653
4654 #[test]
4655 fn test_union_decode_negative_branch_index_errors() {
4656 let union_dt = make_dense_union_avro(
4657 vec![
4658 (Codec::Int32, "i", DataType::Int32),
4659 (Codec::Utf8, "s", DataType::Utf8),
4660 ],
4661 vec![0, 1],
4662 );
4663 let mut dec = Decoder::try_new(&union_dt).unwrap();
4664 let row = encode_avro_long(-1); let err = dec
4666 .decode(&mut AvroCursor::new(&row))
4667 .expect_err("expected error for negative branch index");
4668 let msg = err.to_string();
4669 assert!(
4670 msg.contains("Negative union branch index"),
4671 "unexpected error message: {msg}"
4672 );
4673 }
4674
4675 #[test]
4676 fn test_union_decode_out_of_range_branch_index_errors() {
4677 let union_dt = make_dense_union_avro(
4678 vec![
4679 (Codec::Int32, "i", DataType::Int32),
4680 (Codec::Utf8, "s", DataType::Utf8),
4681 ],
4682 vec![10, 11],
4683 );
4684 let mut dec = Decoder::try_new(&union_dt).unwrap();
4685 let row = encode_avro_long(2);
4686 let err = dec
4687 .decode(&mut AvroCursor::new(&row))
4688 .expect_err("expected error for out-of-range branch index");
4689 let msg = err.to_string();
4690 assert!(
4691 msg.contains("out of range"),
4692 "unexpected error message: {msg}"
4693 );
4694 }
4695
4696 #[test]
4697 fn test_union_sparse_mode_not_supported() {
4698 let children: Vec<AvroDataType> = vec![
4699 AvroDataType::new(Codec::Int32, Default::default(), None),
4700 AvroDataType::new(Codec::Utf8, Default::default(), None),
4701 ];
4702 let uf = UnionFields::try_new(
4703 vec![1, 3],
4704 vec![
4705 arrow_schema::Field::new("i", DataType::Int32, true),
4706 arrow_schema::Field::new("s", DataType::Utf8, true),
4707 ],
4708 )
4709 .unwrap();
4710 let codec = Codec::Union(children.into(), uf, UnionMode::Sparse);
4711 let dt = AvroDataType::new(codec, Default::default(), None);
4712 let err = Decoder::try_new(&dt).expect_err("sparse union should not be supported");
4713 let msg = err.to_string();
4714 assert!(
4715 msg.contains("Sparse Arrow unions are not yet supported"),
4716 "unexpected error message: {msg}"
4717 );
4718 }
4719
4720 fn make_record_decoder_with_projector_defaults(
4721 reader_fields: &[(&str, DataType, bool)],
4722 field_defaults: Vec<Option<AvroLiteral>>,
4723 default_injections: Vec<(usize, AvroLiteral)>,
4724 ) -> Decoder {
4725 assert_eq!(
4726 field_defaults.len(),
4727 reader_fields.len(),
4728 "field_defaults must have one entry per reader field"
4729 );
4730 let mut field_refs: Vec<FieldRef> = Vec::with_capacity(reader_fields.len());
4731 let mut encodings: Vec<Decoder> = Vec::with_capacity(reader_fields.len());
4732 for (name, dt, nullable) in reader_fields {
4733 field_refs.push(Arc::new(ArrowField::new(*name, dt.clone(), *nullable)));
4734 let enc = match dt {
4735 DataType::Int32 => Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY)),
4736 DataType::Int64 => Decoder::Int64(Vec::with_capacity(DEFAULT_CAPACITY)),
4737 DataType::Utf8 => Decoder::String(
4738 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4739 Vec::with_capacity(DEFAULT_CAPACITY),
4740 ),
4741 other => panic!("Unsupported test field type in helper: {other:?}"),
4742 };
4743 encodings.push(enc);
4744 }
4745 let fields: Fields = field_refs.into();
4746 let projector = Projector {
4747 writer_projections: vec![],
4748 default_injections: Arc::from(default_injections),
4749 };
4750 Decoder::Record {
4751 fields,
4752 decoders: encodings,
4753 defaults: field_defaults,
4754 projector: Some(projector),
4755 row_count: 0,
4756 }
4757 }
4758
4759 #[cfg(feature = "avro_custom_types")]
4760 #[test]
4761 fn test_default_append_custom_integer_range_validation() {
4762 let mut d_i8 = Decoder::Int8(Vec::with_capacity(DEFAULT_CAPACITY));
4763 d_i8.append_default(&AvroLiteral::Int(i8::MIN as i32))
4764 .unwrap();
4765 d_i8.append_default(&AvroLiteral::Int(i8::MAX as i32))
4766 .unwrap();
4767 let err_i8_high = d_i8
4768 .append_default(&AvroLiteral::Int(i8::MAX as i32 + 1))
4769 .unwrap_err();
4770 assert!(err_i8_high.to_string().contains("out of range for i8"));
4771 let err_i8_low = d_i8
4772 .append_default(&AvroLiteral::Int(i8::MIN as i32 - 1))
4773 .unwrap_err();
4774 assert!(err_i8_low.to_string().contains("out of range for i8"));
4775 let arr_i8 = d_i8.flush(None).unwrap();
4776 let values_i8 = arr_i8.as_any().downcast_ref::<Int8Array>().unwrap();
4777 assert_eq!(values_i8.values(), &[i8::MIN, i8::MAX]);
4778
4779 let mut d_i16 = Decoder::Int16(Vec::with_capacity(DEFAULT_CAPACITY));
4780 d_i16
4781 .append_default(&AvroLiteral::Int(i16::MIN as i32))
4782 .unwrap();
4783 d_i16
4784 .append_default(&AvroLiteral::Int(i16::MAX as i32))
4785 .unwrap();
4786 let err_i16_high = d_i16
4787 .append_default(&AvroLiteral::Int(i16::MAX as i32 + 1))
4788 .unwrap_err();
4789 assert!(err_i16_high.to_string().contains("out of range for i16"));
4790 let err_i16_low = d_i16
4791 .append_default(&AvroLiteral::Int(i16::MIN as i32 - 1))
4792 .unwrap_err();
4793 assert!(err_i16_low.to_string().contains("out of range for i16"));
4794 let arr_i16 = d_i16.flush(None).unwrap();
4795 let values_i16 = arr_i16.as_any().downcast_ref::<Int16Array>().unwrap();
4796 assert_eq!(values_i16.values(), &[i16::MIN, i16::MAX]);
4797
4798 let mut d_u8 = Decoder::UInt8(Vec::with_capacity(DEFAULT_CAPACITY));
4799 d_u8.append_default(&AvroLiteral::Int(0)).unwrap();
4800 d_u8.append_default(&AvroLiteral::Int(u8::MAX as i32))
4801 .unwrap();
4802 let err_u8_neg = d_u8.append_default(&AvroLiteral::Int(-1)).unwrap_err();
4803 assert!(err_u8_neg.to_string().contains("out of range for u8"));
4804 let err_u8_high = d_u8
4805 .append_default(&AvroLiteral::Int(u8::MAX as i32 + 1))
4806 .unwrap_err();
4807 assert!(err_u8_high.to_string().contains("out of range for u8"));
4808 let arr_u8 = d_u8.flush(None).unwrap();
4809 let values_u8 = arr_u8.as_any().downcast_ref::<UInt8Array>().unwrap();
4810 assert_eq!(values_u8.values(), &[0, u8::MAX]);
4811
4812 let mut d_u16 = Decoder::UInt16(Vec::with_capacity(DEFAULT_CAPACITY));
4813 d_u16.append_default(&AvroLiteral::Int(0)).unwrap();
4814 d_u16
4815 .append_default(&AvroLiteral::Int(u16::MAX as i32))
4816 .unwrap();
4817 let err_u16_neg = d_u16.append_default(&AvroLiteral::Int(-1)).unwrap_err();
4818 assert!(err_u16_neg.to_string().contains("out of range for u16"));
4819 let err_u16_high = d_u16
4820 .append_default(&AvroLiteral::Int(u16::MAX as i32 + 1))
4821 .unwrap_err();
4822 assert!(err_u16_high.to_string().contains("out of range for u16"));
4823 let arr_u16 = d_u16.flush(None).unwrap();
4824 let values_u16 = arr_u16.as_any().downcast_ref::<UInt16Array>().unwrap();
4825 assert_eq!(values_u16.values(), &[0, u16::MAX]);
4826
4827 let mut d_u32 = Decoder::UInt32(Vec::with_capacity(DEFAULT_CAPACITY));
4828 d_u32.append_default(&AvroLiteral::Long(0)).unwrap();
4829 d_u32
4830 .append_default(&AvroLiteral::Long(u32::MAX as i64))
4831 .unwrap();
4832 let err_u32_neg = d_u32.append_default(&AvroLiteral::Long(-1)).unwrap_err();
4833 assert!(err_u32_neg.to_string().contains("out of range for u32"));
4834 let err_u32_high = d_u32
4835 .append_default(&AvroLiteral::Long(u32::MAX as i64 + 1))
4836 .unwrap_err();
4837 assert!(err_u32_high.to_string().contains("out of range for u32"));
4838 let arr_u32 = d_u32.flush(None).unwrap();
4839 let values_u32 = arr_u32.as_any().downcast_ref::<UInt32Array>().unwrap();
4840 assert_eq!(values_u32.values(), &[0, u32::MAX]);
4841 }
4842
4843 #[cfg(feature = "avro_custom_types")]
4844 #[test]
4845 fn test_decode_custom_integer_range_validation() {
4846 let mut d_i8 = Decoder::try_new(&avro_from_codec(Codec::Int8)).unwrap();
4847 d_i8.decode(&mut AvroCursor::new(&encode_avro_int(i8::MIN as i32)))
4848 .unwrap();
4849 d_i8.decode(&mut AvroCursor::new(&encode_avro_int(i8::MAX as i32)))
4850 .unwrap();
4851 let err_i8_high = d_i8
4852 .decode(&mut AvroCursor::new(&encode_avro_int(i8::MAX as i32 + 1)))
4853 .unwrap_err();
4854 assert!(err_i8_high.to_string().contains("out of range for i8"));
4855 let err_i8_low = d_i8
4856 .decode(&mut AvroCursor::new(&encode_avro_int(i8::MIN as i32 - 1)))
4857 .unwrap_err();
4858 assert!(err_i8_low.to_string().contains("out of range for i8"));
4859 let arr_i8 = d_i8.flush(None).unwrap();
4860 let values_i8 = arr_i8.as_any().downcast_ref::<Int8Array>().unwrap();
4861 assert_eq!(values_i8.values(), &[i8::MIN, i8::MAX]);
4862
4863 let mut d_i16 = Decoder::try_new(&avro_from_codec(Codec::Int16)).unwrap();
4864 d_i16
4865 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MIN as i32)))
4866 .unwrap();
4867 d_i16
4868 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MAX as i32)))
4869 .unwrap();
4870 let err_i16_high = d_i16
4871 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MAX as i32 + 1)))
4872 .unwrap_err();
4873 assert!(err_i16_high.to_string().contains("out of range for i16"));
4874 let err_i16_low = d_i16
4875 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MIN as i32 - 1)))
4876 .unwrap_err();
4877 assert!(err_i16_low.to_string().contains("out of range for i16"));
4878 let arr_i16 = d_i16.flush(None).unwrap();
4879 let values_i16 = arr_i16.as_any().downcast_ref::<Int16Array>().unwrap();
4880 assert_eq!(values_i16.values(), &[i16::MIN, i16::MAX]);
4881
4882 let mut d_u8 = Decoder::try_new(&avro_from_codec(Codec::UInt8)).unwrap();
4883 d_u8.decode(&mut AvroCursor::new(&encode_avro_int(0)))
4884 .unwrap();
4885 d_u8.decode(&mut AvroCursor::new(&encode_avro_int(u8::MAX as i32)))
4886 .unwrap();
4887 let err_u8_neg = d_u8
4888 .decode(&mut AvroCursor::new(&encode_avro_int(-1)))
4889 .unwrap_err();
4890 assert!(err_u8_neg.to_string().contains("out of range for u8"));
4891 let err_u8_high = d_u8
4892 .decode(&mut AvroCursor::new(&encode_avro_int(u8::MAX as i32 + 1)))
4893 .unwrap_err();
4894 assert!(err_u8_high.to_string().contains("out of range for u8"));
4895 let arr_u8 = d_u8.flush(None).unwrap();
4896 let values_u8 = arr_u8.as_any().downcast_ref::<UInt8Array>().unwrap();
4897 assert_eq!(values_u8.values(), &[0, u8::MAX]);
4898
4899 let mut d_u16 = Decoder::try_new(&avro_from_codec(Codec::UInt16)).unwrap();
4900 d_u16
4901 .decode(&mut AvroCursor::new(&encode_avro_int(0)))
4902 .unwrap();
4903 d_u16
4904 .decode(&mut AvroCursor::new(&encode_avro_int(u16::MAX as i32)))
4905 .unwrap();
4906 let err_u16_neg = d_u16
4907 .decode(&mut AvroCursor::new(&encode_avro_int(-1)))
4908 .unwrap_err();
4909 assert!(err_u16_neg.to_string().contains("out of range for u16"));
4910 let err_u16_high = d_u16
4911 .decode(&mut AvroCursor::new(&encode_avro_int(u16::MAX as i32 + 1)))
4912 .unwrap_err();
4913 assert!(err_u16_high.to_string().contains("out of range for u16"));
4914 let arr_u16 = d_u16.flush(None).unwrap();
4915 let values_u16 = arr_u16.as_any().downcast_ref::<UInt16Array>().unwrap();
4916 assert_eq!(values_u16.values(), &[0, u16::MAX]);
4917
4918 let mut d_u32 = Decoder::try_new(&avro_from_codec(Codec::UInt32)).unwrap();
4919 d_u32
4920 .decode(&mut AvroCursor::new(&encode_avro_long(0)))
4921 .unwrap();
4922 d_u32
4923 .decode(&mut AvroCursor::new(&encode_avro_long(u32::MAX as i64)))
4924 .unwrap();
4925 let err_u32_neg = d_u32
4926 .decode(&mut AvroCursor::new(&encode_avro_long(-1)))
4927 .unwrap_err();
4928 assert!(err_u32_neg.to_string().contains("out of range for u32"));
4929 let err_u32_high = d_u32
4930 .decode(&mut AvroCursor::new(&encode_avro_long(u32::MAX as i64 + 1)))
4931 .unwrap_err();
4932 assert!(err_u32_high.to_string().contains("out of range for u32"));
4933 let arr_u32 = d_u32.flush(None).unwrap();
4934 let values_u32 = arr_u32.as_any().downcast_ref::<UInt32Array>().unwrap();
4935 assert_eq!(values_u32.values(), &[0, u32::MAX]);
4936 }
4937
4938 #[test]
4939 fn test_default_append_int32_and_int64_from_int_and_long() {
4940 let mut d_i32 = Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY));
4941 d_i32.append_default(&AvroLiteral::Int(42)).unwrap();
4942 let arr = d_i32.flush(None).unwrap();
4943 let a = arr.as_any().downcast_ref::<Int32Array>().unwrap();
4944 assert_eq!(a.len(), 1);
4945 assert_eq!(a.value(0), 42);
4946 let mut d_i64 = Decoder::Int64(Vec::with_capacity(DEFAULT_CAPACITY));
4947 d_i64.append_default(&AvroLiteral::Int(5)).unwrap();
4948 d_i64.append_default(&AvroLiteral::Long(7)).unwrap();
4949 let arr64 = d_i64.flush(None).unwrap();
4950 let a64 = arr64.as_any().downcast_ref::<Int64Array>().unwrap();
4951 assert_eq!(a64.len(), 2);
4952 assert_eq!(a64.value(0), 5);
4953 assert_eq!(a64.value(1), 7);
4954 }
4955
4956 #[test]
4957 fn test_default_append_floats_and_doubles() {
4958 let mut d_f32 = Decoder::Float32(Vec::with_capacity(DEFAULT_CAPACITY));
4959 d_f32.append_default(&AvroLiteral::Float(1.5)).unwrap();
4960 let arr32 = d_f32.flush(None).unwrap();
4961 let a = arr32.as_any().downcast_ref::<Float32Array>().unwrap();
4962 assert_eq!(a.value(0), 1.5);
4963 let mut d_f64 = Decoder::Float64(Vec::with_capacity(DEFAULT_CAPACITY));
4964 d_f64.append_default(&AvroLiteral::Double(2.25)).unwrap();
4965 let arr64 = d_f64.flush(None).unwrap();
4966 let b = arr64.as_any().downcast_ref::<Float64Array>().unwrap();
4967 assert_eq!(b.value(0), 2.25);
4968 }
4969
4970 #[test]
4971 fn test_default_append_string_and_bytes() {
4972 let mut d_str = Decoder::String(
4973 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4974 Vec::with_capacity(DEFAULT_CAPACITY),
4975 );
4976 d_str
4977 .append_default(&AvroLiteral::String("hi".into()))
4978 .unwrap();
4979 let s_arr = d_str.flush(None).unwrap();
4980 let arr = s_arr.as_any().downcast_ref::<StringArray>().unwrap();
4981 assert_eq!(arr.value(0), "hi");
4982 let mut d_bytes = Decoder::Binary(
4983 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4984 Vec::with_capacity(DEFAULT_CAPACITY),
4985 );
4986 d_bytes
4987 .append_default(&AvroLiteral::Bytes(vec![1, 2, 3]))
4988 .unwrap();
4989 let b_arr = d_bytes.flush(None).unwrap();
4990 let barr = b_arr.as_any().downcast_ref::<BinaryArray>().unwrap();
4991 assert_eq!(barr.value(0), &[1, 2, 3]);
4992 let mut d_str_err = Decoder::String(
4993 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4994 Vec::with_capacity(DEFAULT_CAPACITY),
4995 );
4996 let err = d_str_err
4997 .append_default(&AvroLiteral::Bytes(vec![0x61, 0x62]))
4998 .unwrap_err();
4999 assert!(
5000 err.to_string()
5001 .contains("Default for string must be string"),
5002 "unexpected error: {err:?}"
5003 );
5004 }
5005
5006 #[test]
5007 fn test_nullable_null_runs_defer_validity_and_values_together() {
5008 let mut decoder = Decoder::Nullable(NullableDecoder::new(
5009 NullablePlan::ReadTag {
5010 nullability: Nullability::NullFirst,
5011 resolution: ResolutionPlan::Promotion(Promotion::Direct),
5012 },
5013 Decoder::Int32(Vec::new()),
5014 ));
5015
5016 decoder.append_nulls(64).unwrap();
5017 let Decoder::Nullable(nullable) = &decoder else {
5018 unreachable!();
5019 };
5020 assert_eq!(nullable.pending_nulls, 64);
5021 assert_eq!(nullable.validity.len(), 0);
5022 let Decoder::Int32(values) = nullable.values.as_ref() else {
5023 unreachable!();
5024 };
5025 assert!(values.is_empty());
5026
5027 decoder.append_default(&AvroLiteral::Int(7)).unwrap();
5028 let Decoder::Nullable(nullable) = &decoder else {
5029 unreachable!();
5030 };
5031 assert_eq!(nullable.pending_nulls, 0);
5032 assert_eq!(nullable.validity.len(), 65);
5033 let Decoder::Int32(values) = nullable.values.as_ref() else {
5034 unreachable!();
5035 };
5036 assert_eq!(values.len(), 65);
5037 assert_eq!(values[64], 7);
5038
5039 decoder.append_nulls(32).unwrap();
5040 let values = decoder.flush(None).unwrap();
5041 assert_eq!(values.len(), 97);
5042 assert_eq!(values.null_count(), 96);
5043 assert_eq!(values.as_primitive::<Int32Type>().value(64), 7);
5044 }
5045
5046 #[test]
5047 fn test_default_append_nullable_int32_null_and_value() {
5048 let inner = Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY));
5049 let mut dec = Decoder::Nullable(NullableDecoder::new(
5050 NullablePlan::ReadTag {
5051 nullability: Nullability::NullFirst,
5052 resolution: ResolutionPlan::Promotion(Promotion::Direct),
5053 },
5054 inner,
5055 ));
5056 dec.append_default(&AvroLiteral::Null).unwrap();
5057 dec.append_default(&AvroLiteral::Null).unwrap();
5058 dec.append_default(&AvroLiteral::Int(11)).unwrap();
5059 dec.append_default(&AvroLiteral::Null).unwrap();
5060 dec.append_default(&AvroLiteral::Null).unwrap();
5061 dec.append_default(&AvroLiteral::Int(12)).unwrap();
5062 let arr = dec.flush(None).unwrap();
5063 let a = arr.as_any().downcast_ref::<Int32Array>().unwrap();
5064 assert_eq!(a.len(), 6);
5065 assert!(a.is_null(0));
5066 assert!(a.is_null(1));
5067 assert_eq!(a.value(2), 11);
5068 assert!(a.is_null(3));
5069 assert!(a.is_null(4));
5070 assert_eq!(a.value(5), 12);
5071 }
5072
5073 #[test]
5074 fn test_default_append_array_of_ints() {
5075 let list_dt = avro_from_codec(Codec::List(Arc::new(avro_from_codec(Codec::Int32))));
5076 let mut d = Decoder::try_new(&list_dt).unwrap();
5077 let items = vec![
5078 AvroLiteral::Int(1),
5079 AvroLiteral::Int(2),
5080 AvroLiteral::Int(3),
5081 ];
5082 d.append_default(&AvroLiteral::Array(items)).unwrap();
5083 let arr = d.flush(None).unwrap();
5084 let list = arr.as_any().downcast_ref::<ListArray>().unwrap();
5085 assert_eq!(list.len(), 1);
5086 assert_eq!(list.value_length(0), 3);
5087 let vals = list.values().as_any().downcast_ref::<Int32Array>().unwrap();
5088 assert_eq!(vals.values(), &[1, 2, 3]);
5089 }
5090
5091 #[test]
5092 fn test_default_append_map_string_to_int() {
5093 let map_dt = avro_from_codec(Codec::Map(Arc::new(avro_from_codec(Codec::Int32))));
5094 let mut d = Decoder::try_new(&map_dt).unwrap();
5095 let mut m: IndexMap<String, AvroLiteral> = IndexMap::new();
5096 m.insert("k1".to_string(), AvroLiteral::Int(10));
5097 m.insert("k2".to_string(), AvroLiteral::Int(20));
5098 d.append_default(&AvroLiteral::Map(m)).unwrap();
5099 let arr = d.flush(None).unwrap();
5100 let map = arr.as_any().downcast_ref::<MapArray>().unwrap();
5101 assert_eq!(map.len(), 1);
5102 assert_eq!(map.value_length(0), 2);
5103 let binding = map.value(0);
5104 let entries = binding.as_any().downcast_ref::<StructArray>().unwrap();
5105 let k = entries
5106 .column_by_name("key")
5107 .unwrap()
5108 .as_any()
5109 .downcast_ref::<StringArray>()
5110 .unwrap();
5111 let v = entries
5112 .column_by_name("value")
5113 .unwrap()
5114 .as_any()
5115 .downcast_ref::<Int32Array>()
5116 .unwrap();
5117 let keys: std::collections::HashSet<&str> = (0..k.len()).map(|i| k.value(i)).collect();
5118 assert_eq!(keys, ["k1", "k2"].into_iter().collect());
5119 let vals: std::collections::HashSet<i32> = (0..v.len()).map(|i| v.value(i)).collect();
5120 assert_eq!(vals, [10, 20].into_iter().collect());
5121 }
5122
5123 #[test]
5124 fn test_default_append_enum_by_symbol() {
5125 let symbols: Arc<[String]> = vec!["A".into(), "B".into(), "C".into()].into();
5126 let mut d = Decoder::Enum(Vec::with_capacity(DEFAULT_CAPACITY), symbols.clone(), None);
5127 d.append_default(&AvroLiteral::Enum("B".into())).unwrap();
5128 let arr = d.flush(None).unwrap();
5129 let dict = arr
5130 .as_any()
5131 .downcast_ref::<DictionaryArray<Int32Type>>()
5132 .unwrap();
5133 assert_eq!(dict.len(), 1);
5134 let expected = Int32Array::from(vec![1]);
5135 assert_eq!(dict.keys(), &expected);
5136 let values = dict
5137 .values()
5138 .as_any()
5139 .downcast_ref::<StringArray>()
5140 .unwrap();
5141 assert_eq!(values.value(1), "B");
5142 }
5143
5144 #[test]
5145 fn test_default_append_uuid_and_type_error() {
5146 let mut d = Decoder::Uuid(Vec::with_capacity(DEFAULT_CAPACITY));
5147 let uuid_str = "123e4567-e89b-12d3-a456-426614174000";
5148 d.append_default(&AvroLiteral::String(uuid_str.into()))
5149 .unwrap();
5150 let arr_ref = d.flush(None).unwrap();
5151 let arr = arr_ref
5152 .as_any()
5153 .downcast_ref::<FixedSizeBinaryArray>()
5154 .unwrap();
5155 assert_eq!(arr.value_length(), 16);
5156 assert_eq!(arr.len(), 1);
5157 let mut d2 = Decoder::Uuid(Vec::with_capacity(DEFAULT_CAPACITY));
5158 let err = d2
5159 .append_default(&AvroLiteral::Bytes(vec![0u8; 16]))
5160 .unwrap_err();
5161 assert!(
5162 err.to_string().contains("Default for uuid must be string"),
5163 "unexpected error: {err:?}"
5164 );
5165 }
5166
5167 #[test]
5168 fn test_default_append_fixed_and_length_mismatch() {
5169 let mut d = Decoder::Fixed(4, Vec::with_capacity(DEFAULT_CAPACITY));
5170 d.append_default(&AvroLiteral::Bytes(vec![1, 2, 3, 4]))
5171 .unwrap();
5172 let arr_ref = d.flush(None).unwrap();
5173 let arr = arr_ref
5174 .as_any()
5175 .downcast_ref::<FixedSizeBinaryArray>()
5176 .unwrap();
5177 assert_eq!(arr.value_length(), 4);
5178 assert_eq!(arr.value(0), &[1, 2, 3, 4]);
5179 let mut d_err = Decoder::Fixed(4, Vec::with_capacity(DEFAULT_CAPACITY));
5180 let err = d_err
5181 .append_default(&AvroLiteral::Bytes(vec![1, 2, 3]))
5182 .unwrap_err();
5183 assert!(
5184 err.to_string().contains("Fixed default length"),
5185 "unexpected error: {err:?}"
5186 );
5187 }
5188
5189 #[test]
5190 fn test_default_append_duration_and_length_validation() {
5191 let dt = avro_from_codec(Codec::Interval);
5192 let mut d = Decoder::try_new(&dt).unwrap();
5193 let mut bytes = Vec::with_capacity(12);
5194 bytes.extend_from_slice(&1u32.to_le_bytes());
5195 bytes.extend_from_slice(&2u32.to_le_bytes());
5196 bytes.extend_from_slice(&3u32.to_le_bytes());
5197 d.append_default(&AvroLiteral::Bytes(bytes)).unwrap();
5198 let arr_ref = d.flush(None).unwrap();
5199 let arr = arr_ref
5200 .as_any()
5201 .downcast_ref::<IntervalMonthDayNanoArray>()
5202 .unwrap();
5203 assert_eq!(arr.len(), 1);
5204 let v = arr.value(0);
5205 assert_eq!(v.months, 1);
5206 assert_eq!(v.days, 2);
5207 assert_eq!(v.nanoseconds, 3_000_000);
5208 let mut d_err = Decoder::try_new(&avro_from_codec(Codec::Interval)).unwrap();
5209 let err = d_err
5210 .append_default(&AvroLiteral::Bytes(vec![0u8; 11]))
5211 .unwrap_err();
5212 assert!(
5213 err.to_string()
5214 .contains("Duration default must be exactly 12 bytes"),
5215 "unexpected error: {err:?}"
5216 );
5217 }
5218
5219 #[test]
5220 fn test_default_append_decimal256_from_bytes() {
5221 let dt = avro_from_codec(Codec::Decimal(50, Some(2), Some(32)));
5222 let mut d = Decoder::try_new(&dt).unwrap();
5223 let pos: [u8; 32] = [
5224 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
5225 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
5226 0x00, 0x00, 0x30, 0x39,
5227 ];
5228 d.append_default(&AvroLiteral::Bytes(pos.to_vec())).unwrap();
5229 let neg: [u8; 32] = [
5230 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
5231 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
5232 0xFF, 0xFF, 0xFF, 0x85,
5233 ];
5234 d.append_default(&AvroLiteral::Bytes(neg.to_vec())).unwrap();
5235 let arr = d.flush(None).unwrap();
5236 let dec = arr.as_any().downcast_ref::<Decimal256Array>().unwrap();
5237 assert_eq!(dec.len(), 2);
5238 assert_eq!(dec.value_as_string(0), "123.45");
5239 assert_eq!(dec.value_as_string(1), "-1.23");
5240 }
5241
5242 #[test]
5243 fn test_record_append_default_map_missing_fields_uses_projector_field_defaults() {
5244 let field_defaults = vec![None, Some(AvroLiteral::String("hi".into()))];
5245 let mut rec = make_record_decoder_with_projector_defaults(
5246 &[("a", DataType::Int32, false), ("b", DataType::Utf8, false)],
5247 field_defaults,
5248 vec![],
5249 );
5250 let mut map: IndexMap<String, AvroLiteral> = IndexMap::new();
5251 map.insert("a".to_string(), AvroLiteral::Int(7));
5252 rec.append_default(&AvroLiteral::Map(map)).unwrap();
5253 let arr = rec.flush(None).unwrap();
5254 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5255 let a = s
5256 .column_by_name("a")
5257 .unwrap()
5258 .as_any()
5259 .downcast_ref::<Int32Array>()
5260 .unwrap();
5261 let b = s
5262 .column_by_name("b")
5263 .unwrap()
5264 .as_any()
5265 .downcast_ref::<StringArray>()
5266 .unwrap();
5267 assert_eq!(a.value(0), 7);
5268 assert_eq!(b.value(0), "hi");
5269 }
5270
5271 #[test]
5272 fn test_record_append_default_null_uses_projector_field_defaults() {
5273 let field_defaults = vec![
5274 Some(AvroLiteral::Int(5)),
5275 Some(AvroLiteral::String("x".into())),
5276 ];
5277 let mut rec = make_record_decoder_with_projector_defaults(
5278 &[("a", DataType::Int32, false), ("b", DataType::Utf8, false)],
5279 field_defaults,
5280 vec![],
5281 );
5282 rec.append_default(&AvroLiteral::Null).unwrap();
5283 let arr = rec.flush(None).unwrap();
5284 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5285 let a = s
5286 .column_by_name("a")
5287 .unwrap()
5288 .as_any()
5289 .downcast_ref::<Int32Array>()
5290 .unwrap();
5291 let b = s
5292 .column_by_name("b")
5293 .unwrap()
5294 .as_any()
5295 .downcast_ref::<StringArray>()
5296 .unwrap();
5297 assert_eq!(a.value(0), 5);
5298 assert_eq!(b.value(0), "x");
5299 }
5300
5301 #[test]
5302 fn test_record_append_default_missing_fields_without_projector_defaults_yields_type_nulls_or_empties()
5303 {
5304 let fields = vec![("a", DataType::Int32, true), ("b", DataType::Utf8, true)];
5305 let mut field_refs: Vec<FieldRef> = Vec::new();
5306 let mut encoders: Vec<Decoder> = Vec::new();
5307 for (name, dt, nullable) in &fields {
5308 field_refs.push(Arc::new(ArrowField::new(*name, dt.clone(), *nullable)));
5309 }
5310 let enc_a = Decoder::Nullable(NullableDecoder::new(
5311 NullablePlan::ReadTag {
5312 nullability: Nullability::NullSecond,
5313 resolution: ResolutionPlan::Promotion(Promotion::Direct),
5314 },
5315 Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY)),
5316 ));
5317 let enc_b = Decoder::Nullable(NullableDecoder::new(
5318 NullablePlan::ReadTag {
5319 nullability: Nullability::NullSecond,
5320 resolution: ResolutionPlan::Promotion(Promotion::Direct),
5321 },
5322 Decoder::String(
5323 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
5324 Vec::with_capacity(DEFAULT_CAPACITY),
5325 ),
5326 ));
5327 encoders.push(enc_a);
5328 encoders.push(enc_b);
5329 let field_defaults = vec![None, None]; let projector = Projector {
5331 writer_projections: vec![],
5332 default_injections: Arc::from(Vec::<(usize, AvroLiteral)>::new()),
5333 };
5334 let mut rec = Decoder::Record {
5335 fields: field_refs.into(),
5336 decoders: encoders,
5337 defaults: field_defaults,
5338 projector: Some(projector),
5339 row_count: 0,
5340 };
5341 let mut map: IndexMap<String, AvroLiteral> = IndexMap::new();
5342 map.insert("a".to_string(), AvroLiteral::Int(9));
5343 rec.append_default(&AvroLiteral::Map(map)).unwrap();
5344 let arr = rec.flush(None).unwrap();
5345 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5346 let a = s
5347 .column_by_name("a")
5348 .unwrap()
5349 .as_any()
5350 .downcast_ref::<Int32Array>()
5351 .unwrap();
5352 let b = s
5353 .column_by_name("b")
5354 .unwrap()
5355 .as_any()
5356 .downcast_ref::<StringArray>()
5357 .unwrap();
5358 assert!(a.is_valid(0));
5359 assert_eq!(a.value(0), 9);
5360 assert!(b.is_null(0));
5361 }
5362
5363 #[test]
5364 fn test_projector_default_injection_when_writer_lacks_fields() {
5365 let defaults = vec![None, None];
5366 let injections = vec![
5367 (0, AvroLiteral::Int(99)),
5368 (1, AvroLiteral::String("alice".into())),
5369 ];
5370 let mut rec = make_record_decoder_with_projector_defaults(
5371 &[
5372 ("id", DataType::Int32, false),
5373 ("name", DataType::Utf8, false),
5374 ],
5375 defaults,
5376 injections,
5377 );
5378 rec.decode(&mut AvroCursor::new(&[])).unwrap();
5379 let arr = rec.flush(None).unwrap();
5380 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5381 let id = s
5382 .column_by_name("id")
5383 .unwrap()
5384 .as_any()
5385 .downcast_ref::<Int32Array>()
5386 .unwrap();
5387 let name = s
5388 .column_by_name("name")
5389 .unwrap()
5390 .as_any()
5391 .downcast_ref::<StringArray>()
5392 .unwrap();
5393 assert_eq!(id.value(0), 99);
5394 assert_eq!(name.value(0), "alice");
5395 }
5396
5397 #[test]
5398 fn union_type_ids_are_not_child_indexes() {
5399 let encodings: Vec<AvroDataType> =
5400 vec![avro_from_codec(Codec::Int32), avro_from_codec(Codec::Utf8)];
5401 let fields: UnionFields = [
5402 (42_i8, Arc::new(ArrowField::new("a", DataType::Int32, true))),
5403 (7_i8, Arc::new(ArrowField::new("b", DataType::Utf8, true))),
5404 ]
5405 .into_iter()
5406 .collect();
5407 let dt = avro_from_codec(Codec::Union(
5408 encodings.into(),
5409 fields.clone(),
5410 UnionMode::Dense,
5411 ));
5412 let mut dec = Decoder::try_new(&dt).expect("decoder");
5413 let mut b1 = encode_avro_long(1);
5414 b1.extend(encode_avro_bytes(b"hi"));
5415 dec.decode(&mut AvroCursor::new(&b1)).expect("decode b1");
5416 let mut b0 = encode_avro_long(0);
5417 b0.extend(encode_avro_int(5));
5418 dec.decode(&mut AvroCursor::new(&b0)).expect("decode b0");
5419 let arr = dec.flush(None).expect("flush");
5420 let ua = arr.as_any().downcast_ref::<UnionArray>().expect("union");
5421 assert_eq!(ua.len(), 2);
5422 assert_eq!(ua.type_id(0), 7, "type id must come from UnionFields");
5423 assert_eq!(ua.type_id(1), 42, "type id must come from UnionFields");
5424 assert_eq!(ua.value_offset(0), 0);
5425 assert_eq!(ua.value_offset(1), 0);
5426 let utf8_child = ua.child(7).as_any().downcast_ref::<StringArray>().unwrap();
5427 assert_eq!(utf8_child.len(), 1);
5428 assert_eq!(utf8_child.value(0), "hi");
5429 let int_child = ua.child(42).as_any().downcast_ref::<Int32Array>().unwrap();
5430 assert_eq!(int_child.len(), 1);
5431 assert_eq!(int_child.value(0), 5);
5432 let type_ids: Vec<i8> = fields.iter().map(|(tid, _)| tid).collect();
5433 assert_eq!(type_ids, vec![42_i8, 7_i8]);
5434 }
5435
5436 #[cfg(feature = "avro_custom_types")]
5437 #[test]
5438 fn skipper_from_avro_maps_custom_duration_variants_to_int64() -> Result<(), AvroError> {
5439 for codec in [
5440 Codec::DurationNanos,
5441 Codec::DurationMicros,
5442 Codec::DurationMillis,
5443 Codec::DurationSeconds,
5444 ] {
5445 let dt = make_avro_dt(codec.clone(), None);
5446 let s = Skipper::from_avro(&dt)?;
5447 match s {
5448 Skipper::Int64 => {}
5449 other => panic!("expected Int64 skipper for {codec:?}, got {other:?}"),
5450 }
5451 }
5452 Ok(())
5453 }
5454
5455 #[cfg(feature = "avro_custom_types")]
5456 #[test]
5457 fn skipper_skip_consumes_one_long_for_custom_durations() -> Result<(), AvroError> {
5458 let values: [i64; 7] = [0, 1, -1, 150, -150, i64::MAX / 3, i64::MIN / 3];
5459 for codec in [
5460 Codec::DurationNanos,
5461 Codec::DurationMicros,
5462 Codec::DurationMillis,
5463 Codec::DurationSeconds,
5464 ] {
5465 let dt = make_avro_dt(codec.clone(), None);
5466 let s = Skipper::from_avro(&dt)?;
5467 for &v in &values {
5468 let bytes = encode_avro_long(v);
5469 let mut cursor = AvroCursor::new(&bytes);
5470 s.skip(&mut cursor)?;
5471 assert_eq!(
5472 cursor.position(),
5473 bytes.len(),
5474 "did not consume all bytes for {codec:?} value {v}"
5475 );
5476 }
5477 }
5478 Ok(())
5479 }
5480
5481 #[cfg(feature = "avro_custom_types")]
5482 #[test]
5483 fn skipper_nullable_custom_duration_respects_null_first() -> Result<(), AvroError> {
5484 let dt = make_avro_dt(Codec::DurationNanos, Some(Nullability::NullFirst));
5485 let s = Skipper::from_avro(&dt)?;
5486 match &s {
5487 Skipper::Nullable(Nullability::NullFirst, inner) => match **inner {
5488 Skipper::Int64 => {}
5489 ref other => panic!("expected inner Int64, got {other:?}"),
5490 },
5491 other => panic!("expected Nullable(NullFirst, Int64), got {other:?}"),
5492 }
5493 {
5494 let buf = encode_vlq_u64(0);
5495 let mut cursor = AvroCursor::new(&buf);
5496 s.skip(&mut cursor)?;
5497 assert_eq!(cursor.position(), 1, "expected to consume only tag=0");
5498 }
5499 {
5500 let mut buf = encode_vlq_u64(1);
5501 buf.extend(encode_avro_long(0));
5502 let mut cursor = AvroCursor::new(&buf);
5503 s.skip(&mut cursor)?;
5504 assert_eq!(cursor.position(), 2, "expected to consume tag=1 + long(0)");
5505 }
5506
5507 Ok(())
5508 }
5509
5510 #[cfg(feature = "avro_custom_types")]
5511 #[test]
5512 fn skipper_nullable_custom_duration_respects_null_second() -> Result<(), AvroError> {
5513 let dt = make_avro_dt(Codec::DurationMicros, Some(Nullability::NullSecond));
5514 let s = Skipper::from_avro(&dt)?;
5515 match &s {
5516 Skipper::Nullable(Nullability::NullSecond, inner) => match **inner {
5517 Skipper::Int64 => {}
5518 ref other => panic!("expected inner Int64, got {other:?}"),
5519 },
5520 other => panic!("expected Nullable(NullSecond, Int64), got {other:?}"),
5521 }
5522 {
5523 let buf = encode_vlq_u64(1);
5524 let mut cursor = AvroCursor::new(&buf);
5525 s.skip(&mut cursor)?;
5526 assert_eq!(cursor.position(), 1, "expected to consume only tag=1");
5527 }
5528 {
5529 let mut buf = encode_vlq_u64(0);
5530 buf.extend(encode_avro_long(-1));
5531 let mut cursor = AvroCursor::new(&buf);
5532 s.skip(&mut cursor)?;
5533 assert_eq!(
5534 cursor.position(),
5535 1 + encode_avro_long(-1).len(),
5536 "expected to consume tag=0 + long(-1)"
5537 );
5538 }
5539 Ok(())
5540 }
5541
5542 #[test]
5543 fn skipper_interval_is_fixed12_and_skips_12_bytes() -> Result<(), AvroError> {
5544 let dt = make_avro_dt(Codec::Interval, None);
5545 let s = Skipper::from_avro(&dt)?;
5546 match s {
5547 Skipper::DurationFixed12 => {}
5548 other => panic!("expected DurationFixed12, got {other:?}"),
5549 }
5550 let payload = vec![0u8; 12];
5551 let mut cursor = AvroCursor::new(&payload);
5552 s.skip(&mut cursor)?;
5553 assert_eq!(cursor.position(), 12, "expected to consume 12 fixed bytes");
5554 Ok(())
5555 }
5556
5557 #[cfg(feature = "avro_custom_types")]
5558 #[test]
5559 fn test_run_end_encoded_width16_int32_basic_grouping() {
5560 use arrow_array::RunArray;
5561 use std::sync::Arc;
5562 let inner = avro_from_codec(Codec::Int32);
5563 let ree = AvroDataType::new(
5564 Codec::RunEndEncoded(Arc::new(inner), 16),
5565 Default::default(),
5566 None,
5567 );
5568 let mut dec = Decoder::try_new(&ree).expect("create REE decoder");
5569 for v in [1, 1, 1, 2, 2, 3, 3, 3, 3] {
5570 let bytes = encode_avro_int(v);
5571 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5572 }
5573 let arr = dec.flush(None).expect("flush");
5574 let ra = arr
5575 .as_any()
5576 .downcast_ref::<RunArray<Int16Type>>()
5577 .expect("RunArray<Int16Type>");
5578 assert_eq!(ra.len(), 9);
5579 assert_eq!(ra.run_ends().values(), &[3, 5, 9]);
5580 let vals = ra
5581 .values()
5582 .as_ref()
5583 .as_any()
5584 .downcast_ref::<Int32Array>()
5585 .expect("values Int32");
5586 assert_eq!(vals.values(), &[1, 2, 3]);
5587 }
5588
5589 #[cfg(feature = "avro_custom_types")]
5590 #[test]
5591 fn test_run_end_encoded_width32_nullable_values_group_nulls() {
5592 use arrow_array::RunArray;
5593 use std::sync::Arc;
5594 let inner = AvroDataType::new(
5595 Codec::Int32,
5596 Default::default(),
5597 Some(Nullability::NullSecond),
5598 );
5599 let ree = AvroDataType::new(
5600 Codec::RunEndEncoded(Arc::new(inner), 32),
5601 Default::default(),
5602 None,
5603 );
5604 let mut dec = Decoder::try_new(&ree).expect("create REE decoder");
5605 let seq: [Option<i32>; 8] = [
5606 None,
5607 None,
5608 Some(7),
5609 Some(7),
5610 Some(7),
5611 None,
5612 Some(5),
5613 Some(5),
5614 ];
5615 for item in seq {
5616 let mut bytes = Vec::new();
5617 match item {
5618 None => bytes.extend_from_slice(&encode_vlq_u64(1)),
5619 Some(v) => {
5620 bytes.extend_from_slice(&encode_vlq_u64(0));
5621 bytes.extend_from_slice(&encode_avro_int(v));
5622 }
5623 }
5624 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5625 }
5626 let arr = dec.flush(None).expect("flush");
5627 let ra = arr
5628 .as_any()
5629 .downcast_ref::<RunArray<Int32Type>>()
5630 .expect("RunArray<Int32Type>");
5631 assert_eq!(ra.len(), 8);
5632 assert_eq!(ra.run_ends().values(), &[2, 5, 6, 8]);
5633 let vals = ra
5634 .values()
5635 .as_ref()
5636 .as_any()
5637 .downcast_ref::<Int32Array>()
5638 .expect("values Int32 (nullable)");
5639 assert_eq!(vals.len(), 4);
5640 assert!(vals.is_null(0));
5641 assert_eq!(vals.value(1), 7);
5642 assert!(vals.is_null(2));
5643 assert_eq!(vals.value(3), 5);
5644 }
5645
5646 #[cfg(feature = "avro_custom_types")]
5647 #[test]
5648 fn test_run_end_encoded_decode_with_promotion_int_to_double_via_nullable_from_single() {
5649 use arrow_array::RunArray;
5650 let inner_values = Decoder::Float64(Vec::with_capacity(DEFAULT_CAPACITY));
5651 let ree = Decoder::RunEndEncoded(
5652 8, 0,
5654 Box::new(inner_values),
5655 );
5656 let mut dec = Decoder::Nullable(NullableDecoder::new(
5657 NullablePlan::FromSingle {
5658 resolution: ResolutionPlan::Promotion(Promotion::IntToDouble),
5659 },
5660 ree,
5661 ));
5662 for v in [1, 1, 2, 2, 2] {
5663 let bytes = encode_avro_int(v);
5664 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5665 }
5666 let arr = dec.flush(None).expect("flush");
5667 let ra = arr
5668 .as_any()
5669 .downcast_ref::<RunArray<Int64Type>>()
5670 .expect("RunArray<Int64Type>");
5671 assert_eq!(ra.len(), 5);
5672 assert_eq!(ra.run_ends().values(), &[2, 5]);
5673 let vals = ra
5674 .values()
5675 .as_ref()
5676 .as_any()
5677 .downcast_ref::<Float64Array>()
5678 .expect("values Float64");
5679 assert_eq!(vals.values(), &[1.0, 2.0]);
5680 }
5681
5682 #[cfg(feature = "avro_custom_types")]
5683 #[test]
5684 fn test_run_end_encoded_unsupported_run_end_width_errors() {
5685 use std::sync::Arc;
5686 let inner = avro_from_codec(Codec::Int32);
5687 let dt = AvroDataType::new(
5688 Codec::RunEndEncoded(Arc::new(inner), 3),
5689 Default::default(),
5690 None,
5691 );
5692 let err = Decoder::try_new(&dt).expect_err("must reject unsupported width");
5693 let msg = err.to_string();
5694 assert!(
5695 msg.contains("Unsupported run-end width")
5696 && msg.contains("16/32/64 bits or 2/4/8 bytes"),
5697 "unexpected error message: {msg}"
5698 );
5699 }
5700
5701 #[cfg(feature = "avro_custom_types")]
5702 #[test]
5703 fn test_run_end_encoded_empty_input_is_empty_runarray() {
5704 use arrow_array::RunArray;
5705 use std::sync::Arc;
5706 let inner = avro_from_codec(Codec::Utf8);
5707 let dt = AvroDataType::new(
5708 Codec::RunEndEncoded(Arc::new(inner), 4),
5709 Default::default(),
5710 None,
5711 );
5712 let mut dec = Decoder::try_new(&dt).expect("create REE decoder");
5713 let arr = dec.flush(None).expect("flush");
5714 let ra = arr
5715 .as_any()
5716 .downcast_ref::<RunArray<Int32Type>>()
5717 .expect("RunArray<Int32Type>");
5718 assert_eq!(ra.len(), 0);
5719 assert_eq!(ra.run_ends().len(), 0);
5720 assert_eq!(ra.values().len(), 0);
5721 }
5722
5723 #[cfg(feature = "avro_custom_types")]
5724 #[test]
5725 fn test_run_end_encoded_strings_grouping_width32_bits() {
5726 use arrow_array::RunArray;
5727 use std::sync::Arc;
5728 let inner = avro_from_codec(Codec::Utf8);
5729 let dt = AvroDataType::new(
5730 Codec::RunEndEncoded(Arc::new(inner), 32),
5731 Default::default(),
5732 None,
5733 );
5734 let mut dec = Decoder::try_new(&dt).expect("create REE decoder");
5735 for s in ["a", "a", "bb", "bb", "bb", "a"] {
5736 let bytes = encode_avro_bytes(s.as_bytes());
5737 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5738 }
5739 let arr = dec.flush(None).expect("flush");
5740 let ra = arr
5741 .as_any()
5742 .downcast_ref::<RunArray<Int32Type>>()
5743 .expect("RunArray<Int32Type>");
5744 assert_eq!(ra.run_ends().values(), &[2, 5, 6]);
5745 let vals = ra
5746 .values()
5747 .as_ref()
5748 .as_any()
5749 .downcast_ref::<StringArray>()
5750 .expect("values String");
5751 assert_eq!(vals.len(), 3);
5752 assert_eq!(vals.value(0), "a");
5753 assert_eq!(vals.value(1), "bb");
5754 assert_eq!(vals.value(2), "a");
5755 }
5756
5757 #[cfg(not(feature = "avro_custom_types"))]
5758 #[test]
5759 fn test_no_custom_types_feature_smoke_decodes_plain_int32() {
5760 let dt = avro_from_codec(Codec::Int32);
5761 let mut dec = Decoder::try_new(&dt).expect("create Int32 decoder");
5762 for v in [1, 2, 3] {
5763 let bytes = encode_avro_int(v);
5764 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5765 }
5766 let arr = dec.flush(None).expect("flush");
5767 let a = arr
5768 .as_any()
5769 .downcast_ref::<Int32Array>()
5770 .expect("Int32Array");
5771 assert_eq!(a.values(), &[1, 2, 3]);
5772 }
5773
5774 #[test]
5775 fn test_timestamp_nanos_decoding_offset_zero() {
5776 let avro_type = avro_from_codec(Codec::TimestampNanos(Some(Tz::OffsetZero)));
5777 let mut decoder = Decoder::try_new(&avro_type).expect("create TimestampNanos decoder");
5778 let mut data = Vec::new();
5779 for v in [0_i64, 1_i64, -1_i64, 1_234_567_890_i64] {
5780 data.extend_from_slice(&encode_avro_long(v));
5781 }
5782 let mut cur = AvroCursor::new(&data);
5783 for _ in 0..4 {
5784 decoder.decode(&mut cur).expect("decode nanos ts");
5785 }
5786 let array = decoder.flush(None).expect("flush nanos ts");
5787 let ts = array
5788 .as_any()
5789 .downcast_ref::<TimestampNanosecondArray>()
5790 .expect("TimestampNanosecondArray");
5791 assert_eq!(ts.values(), &[0, 1, -1, 1_234_567_890]);
5792 match ts.data_type() {
5793 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5794 assert_eq!(tz.as_deref(), Some("+00:00"));
5795 }
5796 other => panic!("expected Timestamp(Nanosecond, Some(\"+00:00\")), got {other:?}"),
5797 }
5798 }
5799
5800 #[test]
5801 fn test_timestamp_nanos_decoding_utc() {
5802 let avro_type = avro_from_codec(Codec::TimestampNanos(Some(Tz::Utc)));
5803 let mut decoder = Decoder::try_new(&avro_type).expect("create TimestampNanos decoder");
5804 let mut data = Vec::new();
5805 for v in [0_i64, 1_i64, -1_i64, 1_234_567_890_i64] {
5806 data.extend_from_slice(&encode_avro_long(v));
5807 }
5808 let mut cur = AvroCursor::new(&data);
5809 for _ in 0..4 {
5810 decoder.decode(&mut cur).expect("decode nanos ts");
5811 }
5812 let array = decoder.flush(None).expect("flush nanos ts");
5813 let ts = array
5814 .as_any()
5815 .downcast_ref::<TimestampNanosecondArray>()
5816 .expect("TimestampNanosecondArray");
5817 assert_eq!(ts.values(), &[0, 1, -1, 1_234_567_890]);
5818 match ts.data_type() {
5819 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5820 assert_eq!(tz.as_deref(), Some("UTC"));
5821 }
5822 other => panic!("expected Timestamp(Nanosecond, Some(\"UTC\")), got {other:?}"),
5823 }
5824 }
5825
5826 #[test]
5827 fn test_timestamp_nanos_decoding_local() {
5828 let avro_type = avro_from_codec(Codec::TimestampNanos(None));
5829 let mut decoder = Decoder::try_new(&avro_type).expect("create TimestampNanos decoder");
5830 let mut data = Vec::new();
5831 for v in [10_i64, 20_i64, -30_i64] {
5832 data.extend_from_slice(&encode_avro_long(v));
5833 }
5834 let mut cur = AvroCursor::new(&data);
5835 for _ in 0..3 {
5836 decoder.decode(&mut cur).expect("decode nanos ts");
5837 }
5838 let array = decoder.flush(None).expect("flush nanos ts");
5839 let ts = array
5840 .as_any()
5841 .downcast_ref::<TimestampNanosecondArray>()
5842 .expect("TimestampNanosecondArray");
5843 assert_eq!(ts.values(), &[10, 20, -30]);
5844 match ts.data_type() {
5845 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5846 assert_eq!(tz.as_deref(), None);
5847 }
5848 other => panic!("expected Timestamp(Nanosecond, None), got {other:?}"),
5849 }
5850 }
5851
5852 #[test]
5853 fn test_timestamp_nanos_decoding_with_nulls() {
5854 let avro_type = AvroDataType::new(
5855 Codec::TimestampNanos(None),
5856 Default::default(),
5857 Some(Nullability::NullFirst),
5858 );
5859 let mut decoder = Decoder::try_new(&avro_type).expect("create nullable TimestampNanos");
5860 let mut data = Vec::new();
5861 data.extend_from_slice(&encode_avro_long(1));
5862 data.extend_from_slice(&encode_avro_long(42));
5863 data.extend_from_slice(&encode_avro_long(0));
5864 data.extend_from_slice(&encode_avro_long(1));
5865 data.extend_from_slice(&encode_avro_long(-7));
5866 let mut cur = AvroCursor::new(&data);
5867 for _ in 0..3 {
5868 decoder.decode(&mut cur).expect("decode nullable nanos ts");
5869 }
5870 let array = decoder.flush(None).expect("flush nullable nanos ts");
5871 let ts = array
5872 .as_any()
5873 .downcast_ref::<TimestampNanosecondArray>()
5874 .expect("TimestampNanosecondArray");
5875 assert_eq!(ts.len(), 3);
5876 assert!(ts.is_valid(0));
5877 assert!(ts.is_null(1));
5878 assert!(ts.is_valid(2));
5879 assert_eq!(ts.value(0), 42);
5880 assert_eq!(ts.value(2), -7);
5881 match ts.data_type() {
5882 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5883 assert_eq!(tz.as_deref(), None);
5884 }
5885 other => panic!("expected Timestamp(Nanosecond, None), got {other:?}"),
5886 }
5887 }
5888}