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