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 {len} bytes cannot be represented in {N} bytes without overflow"
2417 )));
2418 }
2419 if N > 0 {
2420 let first_kept = raw[extra];
2421 let sign_bit_mismatch = ((first_kept ^ sign_byte) & 0x80) != 0;
2422 if sign_bit_mismatch {
2423 return Err(AvroError::ParseError(format!(
2424 "Decimal value with {len} bytes cannot be represented in {N} bytes without overflow"
2425 )));
2426 }
2427 }
2428 out.copy_from_slice(&raw[extra..]);
2429 return Ok(out);
2430 }
2431 out[N - len..].copy_from_slice(raw);
2432 Ok(out)
2433}
2434
2435#[cfg(feature = "avro_custom_types")]
2436#[inline]
2437fn values_equal_at(arr: &dyn Array, i: usize, j: usize) -> bool {
2438 match (arr.is_null(i), arr.is_null(j)) {
2439 (true, true) => true,
2440 (true, false) | (false, true) => false,
2441 (false, false) => {
2442 let a = arr.slice(i, 1);
2443 let b = arr.slice(j, 1);
2444 a == b
2445 }
2446 }
2447}
2448
2449#[derive(Debug)]
2450struct Projector {
2451 writer_projections: Vec<FieldProjection>,
2452 default_injections: Arc<[(usize, AvroLiteral)]>,
2453}
2454
2455#[derive(Debug)]
2456enum FieldProjection {
2457 ToReader(usize),
2458 Skip(Skipper),
2459}
2460
2461#[derive(Debug)]
2462struct ProjectorBuilder<'a> {
2463 rec: &'a ResolvedRecord,
2464 field_defaults: &'a [Option<AvroLiteral>],
2465}
2466
2467impl<'a> ProjectorBuilder<'a> {
2468 #[inline]
2469 fn try_new(rec: &'a ResolvedRecord, field_defaults: &'a [Option<AvroLiteral>]) -> Self {
2470 Self {
2471 rec,
2472 field_defaults,
2473 }
2474 }
2475
2476 #[inline]
2477 fn build(self) -> Result<Projector, AvroError> {
2478 let mut default_injections: Vec<(usize, AvroLiteral)> =
2479 Vec::with_capacity(self.rec.default_fields.len());
2480 for &idx in self.rec.default_fields.as_ref() {
2481 let lit = self
2482 .field_defaults
2483 .get(idx)
2484 .and_then(|lit| lit.clone())
2485 .unwrap_or(AvroLiteral::Null);
2486 default_injections.push((idx, lit));
2487 }
2488 let writer_projections = self
2489 .rec
2490 .writer_fields
2491 .iter()
2492 .map(|field| match field {
2493 ResolvedField::ToReader(index, _) => Ok(FieldProjection::ToReader(*index)),
2494 ResolvedField::Skip(datatype) => {
2495 let skipper = Skipper::from_avro(datatype)?;
2496 Ok(FieldProjection::Skip(skipper))
2497 }
2498 })
2499 .collect::<Result<_, AvroError>>()?;
2500 Ok(Projector {
2501 writer_projections,
2502 default_injections: default_injections.into(),
2503 })
2504 }
2505}
2506
2507impl Projector {
2508 #[inline]
2509 fn project_record(
2510 &self,
2511 buf: &mut AvroCursor<'_>,
2512 encodings: &mut [Decoder],
2513 ) -> Result<(), AvroError> {
2514 for field_proj in &self.writer_projections {
2515 match field_proj {
2516 FieldProjection::ToReader(index) => encodings[*index].decode(buf)?,
2517 FieldProjection::Skip(skipper) => skipper.skip(buf)?,
2518 }
2519 }
2520 for (reader_index, lit) in self.default_injections.as_ref() {
2521 encodings[*reader_index].append_default(lit)?;
2522 }
2523 Ok(())
2524 }
2525}
2526
2527#[derive(Debug)]
2533enum Skipper {
2534 Null,
2535 Boolean,
2536 Int32,
2537 Int64,
2538 Float32,
2539 Float64,
2540 Bytes,
2541 String,
2542 TimeMicros,
2543 TimestampMillis,
2544 TimestampMicros,
2545 TimestampNanos,
2546 Fixed(usize),
2547 Decimal(Option<usize>),
2548 UuidString,
2549 Enum,
2550 DurationFixed12,
2551 List(Box<Skipper>),
2552 Map(Box<Skipper>),
2553 Struct(Vec<Skipper>),
2554 Union(Vec<Skipper>),
2555 Nullable(Nullability, Box<Skipper>),
2556 #[cfg(feature = "avro_custom_types")]
2557 RunEndEncoded(Box<Skipper>),
2558}
2559
2560impl Skipper {
2561 fn from_avro(dt: &AvroDataType) -> Result<Self, AvroError> {
2562 let mut base = match dt.codec() {
2563 Codec::Null => Self::Null,
2564 Codec::Boolean => Self::Boolean,
2565 Codec::Int32 | Codec::Date32 | Codec::TimeMillis => Self::Int32,
2566 Codec::Int64 => Self::Int64,
2567 Codec::TimeMicros => Self::TimeMicros,
2568 Codec::TimestampMillis(_) => Self::TimestampMillis,
2569 Codec::TimestampMicros(_) => Self::TimestampMicros,
2570 Codec::TimestampNanos(_) => Self::TimestampNanos,
2571 #[cfg(feature = "avro_custom_types")]
2572 Codec::DurationNanos
2573 | Codec::DurationMicros
2574 | Codec::DurationMillis
2575 | Codec::DurationSeconds => Self::Int64,
2576 #[cfg(feature = "avro_custom_types")]
2577 Codec::Int8 | Codec::Int16 | Codec::UInt8 | Codec::UInt16 | Codec::Time32Secs => {
2578 Self::Int32
2579 }
2580 #[cfg(feature = "avro_custom_types")]
2581 Codec::UInt32 | Codec::Date64 | Codec::TimeNanos | Codec::TimestampSecs(_) => {
2582 Self::Int64
2583 }
2584 #[cfg(feature = "avro_custom_types")]
2585 Codec::UInt64 => Self::Fixed(8),
2586 #[cfg(feature = "avro_custom_types")]
2587 Codec::Float16 => Self::Fixed(2),
2588 #[cfg(feature = "avro_custom_types")]
2589 Codec::IntervalYearMonth => Self::Fixed(4),
2590 #[cfg(feature = "avro_custom_types")]
2591 Codec::IntervalMonthDayNano => Self::Fixed(16),
2592 #[cfg(feature = "avro_custom_types")]
2593 Codec::IntervalDayTime => Self::Fixed(8),
2594 Codec::Float32 => Self::Float32,
2595 Codec::Float64 => Self::Float64,
2596 Codec::Binary => Self::Bytes,
2597 Codec::Utf8 | Codec::Utf8View => Self::String,
2598 Codec::Fixed(sz) => Self::Fixed(*sz as usize),
2599 Codec::Decimal(_, _, size) => Self::Decimal(*size),
2600 Codec::Uuid => Self::UuidString, Codec::Enum(_) => Self::Enum,
2602 Codec::List(item) => Self::List(Box::new(Skipper::from_avro(item)?)),
2603 Codec::Struct(fields) => {
2604 if let Some(ResolutionInfo::Record(rec)) = dt.resolution.as_ref() {
2605 Self::Struct(
2606 rec.writer_fields
2607 .iter()
2608 .map(|wf| match wf {
2609 ResolvedField::ToReader(_, wdt) | ResolvedField::Skip(wdt) => {
2610 Skipper::from_avro(wdt)
2611 }
2612 })
2613 .collect::<Result<_, _>>()?,
2614 )
2615 } else {
2616 Self::Struct(
2617 fields
2618 .iter()
2619 .map(|f| Skipper::from_avro(f.data_type()))
2620 .collect::<Result<_, _>>()?,
2621 )
2622 }
2623 }
2624 Codec::Map(values) => Self::Map(Box::new(Skipper::from_avro(values)?)),
2625 Codec::Interval => Self::DurationFixed12,
2626 Codec::Union(encodings, _, _) => {
2627 let max_addr = (i32::MAX as usize) + 1;
2628 if encodings.len() > max_addr {
2629 return Err(AvroError::SchemaError(format!(
2630 "Writer union has {} branches, which exceeds the maximum addressable \
2631 branches by an Avro int tag ({} + 1).",
2632 encodings.len(),
2633 i32::MAX
2634 )));
2635 }
2636 Self::Union(
2637 encodings
2638 .iter()
2639 .map(Skipper::from_avro)
2640 .collect::<Result<_, _>>()?,
2641 )
2642 }
2643 #[cfg(feature = "avro_custom_types")]
2644 Codec::RunEndEncoded(inner, _w) => {
2645 Self::RunEndEncoded(Box::new(Skipper::from_avro(inner)?))
2646 }
2647 };
2648 if let Some(n) = dt.nullability() {
2649 base = Self::Nullable(n, Box::new(base));
2650 }
2651 Ok(base)
2652 }
2653
2654 fn skip(&self, buf: &mut AvroCursor<'_>) -> Result<(), AvroError> {
2655 match self {
2656 Self::Null => Ok(()),
2657 Self::Boolean => {
2658 buf.get_bool()?;
2659 Ok(())
2660 }
2661 Self::Int32 => {
2662 buf.skip_int()?;
2663 Ok(())
2664 }
2665 Self::Int64
2666 | Self::TimeMicros
2667 | Self::TimestampMillis
2668 | Self::TimestampMicros
2669 | Self::TimestampNanos => {
2670 buf.skip_long()?;
2671 Ok(())
2672 }
2673 Self::Float32 => {
2674 buf.get_float()?;
2675 Ok(())
2676 }
2677 Self::Float64 => {
2678 buf.get_double()?;
2679 Ok(())
2680 }
2681 Self::Bytes | Self::String | Self::UuidString => {
2682 buf.get_bytes()?;
2683 Ok(())
2684 }
2685 Self::Fixed(sz) => {
2686 buf.get_fixed(*sz)?;
2687 Ok(())
2688 }
2689 Self::Decimal(size) => {
2690 if let Some(s) = size {
2691 buf.get_fixed(*s)
2692 } else {
2693 buf.get_bytes()
2694 }?;
2695 Ok(())
2696 }
2697 Self::Enum => {
2698 buf.skip_int()?;
2699 Ok(())
2700 }
2701 Self::DurationFixed12 => {
2702 buf.get_fixed(12)?;
2703 Ok(())
2704 }
2705 Self::List(item) => {
2706 skip_blocks(buf, |c| item.skip(c))?;
2707 Ok(())
2708 }
2709 Self::Map(value) => {
2710 skip_blocks(buf, |c| {
2711 c.get_bytes()?; value.skip(c)
2713 })?;
2714 Ok(())
2715 }
2716 Self::Struct(fields) => {
2717 for f in fields {
2718 f.skip(buf)?
2719 }
2720 Ok(())
2721 }
2722 Self::Union(encodings) => {
2723 let raw = buf.get_long()?;
2725 if raw < 0 {
2726 return Err(AvroError::ParseError(format!(
2727 "Negative union branch index {raw}"
2728 )));
2729 }
2730 let idx: usize = usize::try_from(raw).map_err(|_| {
2731 AvroError::ParseError(format!(
2732 "Union branch index {raw} does not fit into usize on this platform ({}-bit)",
2733 (usize::BITS as usize)
2734 ))
2735 })?;
2736 let Some(encoding) = encodings.get(idx) else {
2737 return Err(AvroError::ParseError(format!(
2738 "Union branch index {idx} out of range for skipper ({} branches)",
2739 encodings.len()
2740 )));
2741 };
2742 encoding.skip(buf)
2743 }
2744 Self::Nullable(order, inner) => {
2745 let branch = buf.read_vlq()?;
2746 let is_not_null = match *order {
2747 Nullability::NullFirst => branch != 0,
2748 Nullability::NullSecond => branch == 0,
2749 };
2750 if is_not_null {
2751 inner.skip(buf)?;
2752 }
2753 Ok(())
2754 }
2755 #[cfg(feature = "avro_custom_types")]
2756 Self::RunEndEncoded(inner) => inner.skip(buf),
2757 }
2758 }
2759}
2760
2761#[cfg(test)]
2762mod tests {
2763 use super::*;
2764 use crate::codec::AvroFieldBuilder;
2765 use crate::schema::{Attributes, ComplexType, Field, PrimitiveType, Record, Schema, TypeName};
2766 use arrow_array::cast::AsArray;
2767 use indexmap::IndexMap;
2768 use std::collections::HashMap;
2769
2770 fn encode_avro_int(value: i32) -> Vec<u8> {
2771 let mut buf = Vec::new();
2772 let mut v = (value << 1) ^ (value >> 31);
2773 while v & !0x7F != 0 {
2774 buf.push(((v & 0x7F) | 0x80) as u8);
2775 v >>= 7;
2776 }
2777 buf.push(v as u8);
2778 buf
2779 }
2780
2781 fn encode_avro_long(value: i64) -> Vec<u8> {
2782 let mut buf = Vec::new();
2783 let mut v = (value << 1) ^ (value >> 63);
2784 while v & !0x7F != 0 {
2785 buf.push(((v & 0x7F) | 0x80) as u8);
2786 v >>= 7;
2787 }
2788 buf.push(v as u8);
2789 buf
2790 }
2791
2792 fn encode_avro_bytes(bytes: &[u8]) -> Vec<u8> {
2793 let mut buf = encode_avro_long(bytes.len() as i64);
2794 buf.extend_from_slice(bytes);
2795 buf
2796 }
2797
2798 fn avro_from_codec(codec: Codec) -> AvroDataType {
2799 AvroDataType::new(codec, Default::default(), None)
2800 }
2801
2802 fn resolved_root_datatype(
2803 writer: Schema<'static>,
2804 reader: Schema<'static>,
2805 use_utf8view: bool,
2806 strict_mode: bool,
2807 ) -> AvroDataType {
2808 let writer_record = Schema::Complex(ComplexType::Record(Record {
2810 name: "Root",
2811 namespace: None,
2812 doc: None,
2813 aliases: vec![],
2814 fields: vec![Field {
2815 name: "v",
2816 r#type: writer,
2817 default: None,
2818 doc: None,
2819 aliases: vec![],
2820 }],
2821 attributes: Attributes::default(),
2822 }));
2823
2824 let reader_record = Schema::Complex(ComplexType::Record(Record {
2826 name: "Root",
2827 namespace: None,
2828 doc: None,
2829 aliases: vec![],
2830 fields: vec![Field {
2831 name: "v",
2832 r#type: reader,
2833 default: None,
2834 doc: None,
2835 aliases: vec![],
2836 }],
2837 attributes: Attributes::default(),
2838 }));
2839
2840 let field = AvroFieldBuilder::new(&writer_record)
2842 .with_reader_schema(&reader_record)
2843 .with_utf8view(use_utf8view)
2844 .with_strict_mode(strict_mode)
2845 .build()
2846 .expect("schema resolution should succeed");
2847
2848 match field.data_type().codec() {
2849 Codec::Struct(fields) => fields[0].data_type().clone(),
2850 other => panic!("expected wrapper struct, got {other:?}"),
2851 }
2852 }
2853
2854 fn decoder_for_promotion(
2855 writer: PrimitiveType,
2856 reader: PrimitiveType,
2857 use_utf8view: bool,
2858 ) -> Decoder {
2859 let ws = Schema::TypeName(TypeName::Primitive(writer));
2860 let rs = Schema::TypeName(TypeName::Primitive(reader));
2861 let dt = resolved_root_datatype(ws, rs, use_utf8view, false);
2862 Decoder::try_new(&dt).unwrap()
2863 }
2864
2865 fn make_avro_dt(codec: Codec, nullability: Option<Nullability>) -> AvroDataType {
2866 AvroDataType::new(codec, HashMap::new(), nullability)
2867 }
2868
2869 #[cfg(feature = "avro_custom_types")]
2870 fn encode_vlq_u64(mut x: u64) -> Vec<u8> {
2871 let mut out = Vec::with_capacity(10);
2872 while x >= 0x80 {
2873 out.push((x as u8) | 0x80);
2874 x >>= 7;
2875 }
2876 out.push(x as u8);
2877 out
2878 }
2879
2880 #[test]
2881 fn test_union_resolution_writer_union_reader_union_reorder_and_promotion_dense() {
2882 let ws = Schema::Union(vec![
2883 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2884 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2885 ]);
2886 let rs = Schema::Union(vec![
2887 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2888 Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
2889 ]);
2890
2891 let dt = resolved_root_datatype(ws, rs, false, false);
2892 let mut dec = Decoder::try_new(&dt).unwrap();
2893
2894 let mut rec1 = encode_avro_long(0);
2895 rec1.extend(encode_avro_int(7));
2896 let mut cur1 = AvroCursor::new(&rec1);
2897 dec.decode(&mut cur1).unwrap();
2898
2899 let mut rec2 = encode_avro_long(1);
2900 rec2.extend(encode_avro_bytes(b"abc"));
2901 let mut cur2 = AvroCursor::new(&rec2);
2902 dec.decode(&mut cur2).unwrap();
2903
2904 let arr = dec.flush(None).unwrap();
2905 let ua = arr
2906 .as_any()
2907 .downcast_ref::<UnionArray>()
2908 .expect("dense union output");
2909
2910 assert_eq!(
2911 ua.type_id(0),
2912 1,
2913 "first value must select reader 'long' branch"
2914 );
2915 assert_eq!(ua.value_offset(0), 0);
2916
2917 assert_eq!(
2918 ua.type_id(1),
2919 0,
2920 "second value must select reader 'string' branch"
2921 );
2922 assert_eq!(ua.value_offset(1), 0);
2923
2924 let long_child = ua.child(1).as_any().downcast_ref::<Int64Array>().unwrap();
2925 assert_eq!(long_child.len(), 1);
2926 assert_eq!(long_child.value(0), 7);
2927
2928 let str_child = ua.child(0).as_any().downcast_ref::<StringArray>().unwrap();
2929 assert_eq!(str_child.len(), 1);
2930 assert_eq!(str_child.value(0), "abc");
2931 }
2932
2933 #[test]
2934 fn test_union_resolution_writer_union_reader_nonunion_promotion_int_to_long() {
2935 let ws = Schema::Union(vec![
2936 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2937 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2938 ]);
2939 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Long));
2940
2941 let dt = resolved_root_datatype(ws, rs, false, false);
2942 let mut dec = Decoder::try_new(&dt).unwrap();
2943
2944 let mut data = encode_avro_long(0);
2945 data.extend(encode_avro_int(5));
2946 let mut cur = AvroCursor::new(&data);
2947 dec.decode(&mut cur).unwrap();
2948
2949 let arr = dec.flush(None).unwrap();
2950 let out = arr.as_any().downcast_ref::<Int64Array>().unwrap();
2951 assert_eq!(out.len(), 1);
2952 assert_eq!(out.value(0), 5);
2953 }
2954
2955 #[test]
2956 fn test_union_resolution_writer_union_reader_nonunion_mismatch_errors() {
2957 let ws = Schema::Union(vec![
2958 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
2959 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2960 ]);
2961 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Long));
2962
2963 let dt = resolved_root_datatype(ws, rs, false, false);
2964 let mut dec = Decoder::try_new(&dt).unwrap();
2965
2966 let mut data = encode_avro_long(1);
2967 data.extend(encode_avro_bytes(b"z"));
2968 let mut cur = AvroCursor::new(&data);
2969 let res = dec.decode(&mut cur);
2970 assert!(
2971 res.is_err(),
2972 "expected error when writer union branch does not resolve to reader non-union type"
2973 );
2974 }
2975
2976 #[test]
2977 fn test_union_resolution_writer_nonunion_reader_union_selects_matching_branch() {
2978 let ws = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
2979 let rs = Schema::Union(vec![
2980 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
2981 Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
2982 ]);
2983
2984 let dt = resolved_root_datatype(ws, rs, false, false);
2985 let mut dec = Decoder::try_new(&dt).unwrap();
2986
2987 let data = encode_avro_int(6);
2988 let mut cur = AvroCursor::new(&data);
2989 dec.decode(&mut cur).unwrap();
2990
2991 let arr = dec.flush(None).unwrap();
2992 let ua = arr
2993 .as_any()
2994 .downcast_ref::<UnionArray>()
2995 .expect("dense union output");
2996 assert_eq!(ua.len(), 1);
2997 assert_eq!(
2998 ua.type_id(0),
2999 1,
3000 "must resolve to reader 'long' branch (type_id 1)"
3001 );
3002 assert_eq!(ua.value_offset(0), 0);
3003
3004 let long_child = ua.child(1).as_any().downcast_ref::<Int64Array>().unwrap();
3005 assert_eq!(long_child.len(), 1);
3006 assert_eq!(long_child.value(0), 6);
3007
3008 let str_child = ua.child(0).as_any().downcast_ref::<StringArray>().unwrap();
3009 assert_eq!(str_child.len(), 0, "string branch must be empty");
3010 }
3011
3012 #[test]
3013 fn test_union_resolution_writer_union_reader_union_unmapped_branch_errors() {
3014 let ws = Schema::Union(vec![
3015 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3016 Schema::TypeName(TypeName::Primitive(PrimitiveType::Boolean)),
3017 ]);
3018 let rs = Schema::Union(vec![
3019 Schema::TypeName(TypeName::Primitive(PrimitiveType::String)),
3020 Schema::TypeName(TypeName::Primitive(PrimitiveType::Long)),
3021 ]);
3022
3023 let dt = resolved_root_datatype(ws, rs, false, false);
3024 let mut dec = Decoder::try_new(&dt).unwrap();
3025
3026 let mut data = encode_avro_long(1);
3027 data.push(1);
3028 let mut cur = AvroCursor::new(&data);
3029 let res = dec.decode(&mut cur);
3030 assert!(
3031 res.is_err(),
3032 "expected error for unmapped writer 'boolean' branch"
3033 );
3034 }
3035
3036 #[test]
3037 fn test_schema_resolution_promotion_int_to_long() {
3038 let mut dec = decoder_for_promotion(PrimitiveType::Int, PrimitiveType::Long, false);
3039 assert!(matches!(dec, Decoder::Int32ToInt64(_)));
3040 for v in [0, 1, -2, 123456] {
3041 let data = encode_avro_int(v);
3042 let mut cur = AvroCursor::new(&data);
3043 dec.decode(&mut cur).unwrap();
3044 }
3045 let arr = dec.flush(None).unwrap();
3046 let a = arr.as_any().downcast_ref::<Int64Array>().unwrap();
3047 assert_eq!(a.value(0), 0);
3048 assert_eq!(a.value(1), 1);
3049 assert_eq!(a.value(2), -2);
3050 assert_eq!(a.value(3), 123456);
3051 }
3052
3053 #[test]
3054 fn test_schema_resolution_promotion_int_to_float() {
3055 let mut dec = decoder_for_promotion(PrimitiveType::Int, PrimitiveType::Float, false);
3056 assert!(matches!(dec, Decoder::Int32ToFloat32(_)));
3057 for v in [0, 42, -7] {
3058 let data = encode_avro_int(v);
3059 let mut cur = AvroCursor::new(&data);
3060 dec.decode(&mut cur).unwrap();
3061 }
3062 let arr = dec.flush(None).unwrap();
3063 let a = arr.as_any().downcast_ref::<Float32Array>().unwrap();
3064 assert_eq!(a.value(0), 0.0);
3065 assert_eq!(a.value(1), 42.0);
3066 assert_eq!(a.value(2), -7.0);
3067 }
3068
3069 #[test]
3070 fn test_schema_resolution_promotion_int_to_double() {
3071 let mut dec = decoder_for_promotion(PrimitiveType::Int, PrimitiveType::Double, false);
3072 assert!(matches!(dec, Decoder::Int32ToFloat64(_)));
3073 for v in [1, -1, 10_000] {
3074 let data = encode_avro_int(v);
3075 let mut cur = AvroCursor::new(&data);
3076 dec.decode(&mut cur).unwrap();
3077 }
3078 let arr = dec.flush(None).unwrap();
3079 let a = arr.as_any().downcast_ref::<Float64Array>().unwrap();
3080 assert_eq!(a.value(0), 1.0);
3081 assert_eq!(a.value(1), -1.0);
3082 assert_eq!(a.value(2), 10_000.0);
3083 }
3084
3085 #[test]
3086 fn test_schema_resolution_promotion_long_to_float() {
3087 let mut dec = decoder_for_promotion(PrimitiveType::Long, PrimitiveType::Float, false);
3088 assert!(matches!(dec, Decoder::Int64ToFloat32(_)));
3089 for v in [0_i64, 1_000_000_i64, -123_i64] {
3090 let data = encode_avro_long(v);
3091 let mut cur = AvroCursor::new(&data);
3092 dec.decode(&mut cur).unwrap();
3093 }
3094 let arr = dec.flush(None).unwrap();
3095 let a = arr.as_any().downcast_ref::<Float32Array>().unwrap();
3096 assert_eq!(a.value(0), 0.0);
3097 assert_eq!(a.value(1), 1_000_000.0);
3098 assert_eq!(a.value(2), -123.0);
3099 }
3100
3101 #[test]
3102 fn test_schema_resolution_promotion_long_to_double() {
3103 let mut dec = decoder_for_promotion(PrimitiveType::Long, PrimitiveType::Double, false);
3104 assert!(matches!(dec, Decoder::Int64ToFloat64(_)));
3105 for v in [2_i64, -2_i64, 9_223_372_i64] {
3106 let data = encode_avro_long(v);
3107 let mut cur = AvroCursor::new(&data);
3108 dec.decode(&mut cur).unwrap();
3109 }
3110 let arr = dec.flush(None).unwrap();
3111 let a = arr.as_any().downcast_ref::<Float64Array>().unwrap();
3112 assert_eq!(a.value(0), 2.0);
3113 assert_eq!(a.value(1), -2.0);
3114 assert_eq!(a.value(2), 9_223_372.0);
3115 }
3116
3117 #[test]
3118 fn test_schema_resolution_promotion_float_to_double() {
3119 let mut dec = decoder_for_promotion(PrimitiveType::Float, PrimitiveType::Double, false);
3120 assert!(matches!(dec, Decoder::Float32ToFloat64(_)));
3121 for v in [0.5_f32, -3.25_f32, 1.0e6_f32] {
3122 let data = v.to_le_bytes().to_vec();
3123 let mut cur = AvroCursor::new(&data);
3124 dec.decode(&mut cur).unwrap();
3125 }
3126 let arr = dec.flush(None).unwrap();
3127 let a = arr.as_any().downcast_ref::<Float64Array>().unwrap();
3128 assert_eq!(a.value(0), 0.5_f64);
3129 assert_eq!(a.value(1), -3.25_f64);
3130 assert_eq!(a.value(2), 1.0e6_f64);
3131 }
3132
3133 #[test]
3134 fn test_schema_resolution_promotion_bytes_to_string_utf8() {
3135 let mut dec = decoder_for_promotion(PrimitiveType::Bytes, PrimitiveType::String, false);
3136 assert!(matches!(dec, Decoder::BytesToString(_, _)));
3137 for s in ["hello", "world", "héllo"] {
3138 let data = encode_avro_bytes(s.as_bytes());
3139 let mut cur = AvroCursor::new(&data);
3140 dec.decode(&mut cur).unwrap();
3141 }
3142 let arr = dec.flush(None).unwrap();
3143 let a = arr.as_any().downcast_ref::<StringArray>().unwrap();
3144 assert_eq!(a.value(0), "hello");
3145 assert_eq!(a.value(1), "world");
3146 assert_eq!(a.value(2), "héllo");
3147 }
3148
3149 #[test]
3150 fn test_schema_resolution_promotion_bytes_to_string_utf8view_enabled() {
3151 let mut dec = decoder_for_promotion(PrimitiveType::Bytes, PrimitiveType::String, true);
3152 assert!(matches!(dec, Decoder::BytesToString(_, _)));
3153 let data = encode_avro_bytes(b"abc");
3154 let mut cur = AvroCursor::new(&data);
3155 dec.decode(&mut cur).unwrap();
3156 let arr = dec.flush(None).unwrap();
3157 let a = arr.as_any().downcast_ref::<StringArray>().unwrap();
3158 assert_eq!(a.value(0), "abc");
3159 }
3160
3161 #[test]
3162 fn test_schema_resolution_promotion_string_to_bytes() {
3163 let mut dec = decoder_for_promotion(PrimitiveType::String, PrimitiveType::Bytes, false);
3164 assert!(matches!(dec, Decoder::StringToBytes(_, _)));
3165 for s in ["", "abc", "data"] {
3166 let data = encode_avro_bytes(s.as_bytes());
3167 let mut cur = AvroCursor::new(&data);
3168 dec.decode(&mut cur).unwrap();
3169 }
3170 let arr = dec.flush(None).unwrap();
3171 let a = arr.as_any().downcast_ref::<BinaryArray>().unwrap();
3172 assert_eq!(a.value(0), b"");
3173 assert_eq!(a.value(1), b"abc");
3174 assert_eq!(a.value(2), b"data");
3175 }
3176
3177 #[test]
3178 fn test_schema_resolution_no_promotion_passthrough_int() {
3179 let ws = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3180 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3181 let writer_record = Schema::Complex(ComplexType::Record(Record {
3183 name: "Root",
3184 namespace: None,
3185 doc: None,
3186 aliases: vec![],
3187 fields: vec![Field {
3188 name: "v",
3189 r#type: ws,
3190 default: None,
3191 doc: None,
3192 aliases: vec![],
3193 }],
3194 attributes: Attributes::default(),
3195 }));
3196 let reader_record = Schema::Complex(ComplexType::Record(Record {
3197 name: "Root",
3198 namespace: None,
3199 doc: None,
3200 aliases: vec![],
3201 fields: vec![Field {
3202 name: "v",
3203 r#type: rs,
3204 default: None,
3205 doc: None,
3206 aliases: vec![],
3207 }],
3208 attributes: Attributes::default(),
3209 }));
3210 let field = AvroFieldBuilder::new(&writer_record)
3211 .with_reader_schema(&reader_record)
3212 .with_utf8view(false)
3213 .with_strict_mode(false)
3214 .build()
3215 .unwrap();
3216 let dt = match field.data_type().codec() {
3218 Codec::Struct(fields) => fields[0].data_type().clone(),
3219 other => panic!("expected wrapper struct, got {other:?}"),
3220 };
3221 let mut dec = Decoder::try_new(&dt).unwrap();
3222 assert!(matches!(dec, Decoder::Int32(_)));
3223 for v in [7, -9] {
3224 let data = encode_avro_int(v);
3225 let mut cur = AvroCursor::new(&data);
3226 dec.decode(&mut cur).unwrap();
3227 }
3228 let arr = dec.flush(None).unwrap();
3229 let a = arr.as_any().downcast_ref::<Int32Array>().unwrap();
3230 assert_eq!(a.value(0), 7);
3231 assert_eq!(a.value(1), -9);
3232 }
3233
3234 #[test]
3235 fn test_schema_resolution_illegal_promotion_int_to_boolean_errors() {
3236 let ws = Schema::TypeName(TypeName::Primitive(PrimitiveType::Int));
3237 let rs = Schema::TypeName(TypeName::Primitive(PrimitiveType::Boolean));
3238 let writer_record = Schema::Complex(ComplexType::Record(Record {
3239 name: "Root",
3240 namespace: None,
3241 doc: None,
3242 aliases: vec![],
3243 fields: vec![Field {
3244 name: "v",
3245 r#type: ws,
3246 default: None,
3247 doc: None,
3248 aliases: vec![],
3249 }],
3250 attributes: Attributes::default(),
3251 }));
3252 let reader_record = Schema::Complex(ComplexType::Record(Record {
3253 name: "Root",
3254 namespace: None,
3255 doc: None,
3256 aliases: vec![],
3257 fields: vec![Field {
3258 name: "v",
3259 r#type: rs,
3260 default: None,
3261 doc: None,
3262 aliases: vec![],
3263 }],
3264 attributes: Attributes::default(),
3265 }));
3266 let res = AvroFieldBuilder::new(&writer_record)
3267 .with_reader_schema(&reader_record)
3268 .with_utf8view(false)
3269 .with_strict_mode(false)
3270 .build();
3271 assert!(res.is_err(), "expected error for illegal promotion");
3272 }
3273
3274 #[test]
3275 fn test_map_decoding_one_entry() {
3276 let value_type = avro_from_codec(Codec::Utf8);
3277 let map_type = avro_from_codec(Codec::Map(Arc::new(value_type)));
3278 let mut decoder = Decoder::try_new(&map_type).unwrap();
3279 let mut data = Vec::new();
3281 data.extend_from_slice(&encode_avro_long(1));
3282 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));
3285 let mut cursor = AvroCursor::new(&data);
3286 decoder.decode(&mut cursor).unwrap();
3287 let array = decoder.flush(None).unwrap();
3288 let map_arr = array.as_any().downcast_ref::<MapArray>().unwrap();
3289 assert_eq!(map_arr.len(), 1); assert_eq!(map_arr.value_length(0), 1);
3291 let entries = map_arr.value(0);
3292 let struct_entries = entries.as_any().downcast_ref::<StructArray>().unwrap();
3293 assert_eq!(struct_entries.len(), 1);
3294 let key_arr = struct_entries
3295 .column_by_name("key")
3296 .unwrap()
3297 .as_any()
3298 .downcast_ref::<StringArray>()
3299 .unwrap();
3300 let val_arr = struct_entries
3301 .column_by_name("value")
3302 .unwrap()
3303 .as_any()
3304 .downcast_ref::<StringArray>()
3305 .unwrap();
3306 assert_eq!(key_arr.value(0), "hello");
3307 assert_eq!(val_arr.value(0), "world");
3308 }
3309
3310 #[test]
3311 fn test_map_decoding_empty() {
3312 let value_type = avro_from_codec(Codec::Utf8);
3313 let map_type = avro_from_codec(Codec::Map(Arc::new(value_type)));
3314 let mut decoder = Decoder::try_new(&map_type).unwrap();
3315 let data = encode_avro_long(0);
3316 decoder.decode(&mut AvroCursor::new(&data)).unwrap();
3317 let array = decoder.flush(None).unwrap();
3318 let map_arr = array.as_any().downcast_ref::<MapArray>().unwrap();
3319 assert_eq!(map_arr.len(), 1);
3320 assert_eq!(map_arr.value_length(0), 0);
3321 }
3322
3323 #[test]
3324 fn test_fixed_decoding() {
3325 let avro_type = avro_from_codec(Codec::Fixed(3));
3326 let mut decoder = Decoder::try_new(&avro_type).expect("Failed to create decoder");
3327
3328 let data1 = [1u8, 2, 3];
3329 let mut cursor1 = AvroCursor::new(&data1);
3330 decoder
3331 .decode(&mut cursor1)
3332 .expect("Failed to decode data1");
3333 assert_eq!(cursor1.position(), 3, "Cursor should advance by fixed size");
3334 let data2 = [4u8, 5, 6];
3335 let mut cursor2 = AvroCursor::new(&data2);
3336 decoder
3337 .decode(&mut cursor2)
3338 .expect("Failed to decode data2");
3339 assert_eq!(cursor2.position(), 3, "Cursor should advance by fixed size");
3340 let array = decoder.flush(None).expect("Failed to flush decoder");
3341 assert_eq!(array.len(), 2, "Array should contain two items");
3342 let fixed_size_binary_array = array
3343 .as_any()
3344 .downcast_ref::<FixedSizeBinaryArray>()
3345 .expect("Failed to downcast to FixedSizeBinaryArray");
3346 assert_eq!(
3347 fixed_size_binary_array.value_length(),
3348 3,
3349 "Fixed size of binary values should be 3"
3350 );
3351 assert_eq!(
3352 fixed_size_binary_array.value(0),
3353 &[1, 2, 3],
3354 "First item mismatch"
3355 );
3356 assert_eq!(
3357 fixed_size_binary_array.value(1),
3358 &[4, 5, 6],
3359 "Second item mismatch"
3360 );
3361 }
3362
3363 #[test]
3364 fn test_fixed_decoding_empty() {
3365 let avro_type = avro_from_codec(Codec::Fixed(5));
3366 let mut decoder = Decoder::try_new(&avro_type).expect("Failed to create decoder");
3367
3368 let array = decoder
3369 .flush(None)
3370 .expect("Failed to flush decoder for empty input");
3371
3372 assert_eq!(array.len(), 0, "Array should be empty");
3373 let fixed_size_binary_array = array
3374 .as_any()
3375 .downcast_ref::<FixedSizeBinaryArray>()
3376 .expect("Failed to downcast to FixedSizeBinaryArray for empty array");
3377
3378 assert_eq!(
3379 fixed_size_binary_array.value_length(),
3380 5,
3381 "Fixed size of binary values should be 5 as per type"
3382 );
3383 }
3384
3385 #[test]
3386 fn test_uuid_decoding() {
3387 let avro_type = avro_from_codec(Codec::Uuid);
3388 let mut decoder = Decoder::try_new(&avro_type).expect("Failed to create decoder");
3389 let uuid_str = "f81d4fae-7dec-11d0-a765-00a0c91e6bf6";
3390 let data = encode_avro_bytes(uuid_str.as_bytes());
3391 let mut cursor = AvroCursor::new(&data);
3392 decoder.decode(&mut cursor).expect("Failed to decode data");
3393 assert_eq!(
3394 cursor.position(),
3395 data.len(),
3396 "Cursor should advance by varint size + data size"
3397 );
3398 let array = decoder.flush(None).expect("Failed to flush decoder");
3399 let fixed_size_binary_array = array
3400 .as_any()
3401 .downcast_ref::<FixedSizeBinaryArray>()
3402 .expect("Array should be a FixedSizeBinaryArray");
3403 assert_eq!(fixed_size_binary_array.len(), 1);
3404 assert_eq!(fixed_size_binary_array.value_length(), 16);
3405 let expected_bytes = [
3406 0xf8, 0x1d, 0x4f, 0xae, 0x7d, 0xec, 0x11, 0xd0, 0xa7, 0x65, 0x00, 0xa0, 0xc9, 0x1e,
3407 0x6b, 0xf6,
3408 ];
3409 assert_eq!(fixed_size_binary_array.value(0), &expected_bytes);
3410 }
3411
3412 #[test]
3413 fn test_array_decoding() {
3414 let item_dt = avro_from_codec(Codec::Int32);
3415 let list_dt = avro_from_codec(Codec::List(Arc::new(item_dt)));
3416 let mut decoder = Decoder::try_new(&list_dt).unwrap();
3417 let mut row1 = Vec::new();
3418 row1.extend_from_slice(&encode_avro_long(2));
3419 row1.extend_from_slice(&encode_avro_int(10));
3420 row1.extend_from_slice(&encode_avro_int(20));
3421 row1.extend_from_slice(&encode_avro_long(0));
3422 let row2 = encode_avro_long(0);
3423 let mut cursor = AvroCursor::new(&row1);
3424 decoder.decode(&mut cursor).unwrap();
3425 let mut cursor2 = AvroCursor::new(&row2);
3426 decoder.decode(&mut cursor2).unwrap();
3427 let array = decoder.flush(None).unwrap();
3428 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3429 assert_eq!(list_arr.len(), 2);
3430 let offsets = list_arr.value_offsets();
3431 assert_eq!(offsets, &[0, 2, 2]);
3432 let values = list_arr.values();
3433 let int_arr = values.as_primitive::<Int32Type>();
3434 assert_eq!(int_arr.len(), 2);
3435 assert_eq!(int_arr.value(0), 10);
3436 assert_eq!(int_arr.value(1), 20);
3437 }
3438
3439 #[test]
3440 fn test_array_decoding_with_negative_block_count() {
3441 let item_dt = avro_from_codec(Codec::Int32);
3442 let list_dt = avro_from_codec(Codec::List(Arc::new(item_dt)));
3443 let mut decoder = Decoder::try_new(&list_dt).unwrap();
3444 let mut data = encode_avro_long(-3);
3445 data.extend_from_slice(&encode_avro_long(12));
3446 data.extend_from_slice(&encode_avro_int(1));
3447 data.extend_from_slice(&encode_avro_int(2));
3448 data.extend_from_slice(&encode_avro_int(3));
3449 data.extend_from_slice(&encode_avro_long(0));
3450 let mut cursor = AvroCursor::new(&data);
3451 decoder.decode(&mut cursor).unwrap();
3452 let array = decoder.flush(None).unwrap();
3453 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3454 assert_eq!(list_arr.len(), 1);
3455 assert_eq!(list_arr.value_length(0), 3);
3456 let values = list_arr.values().as_primitive::<Int32Type>();
3457 assert_eq!(values.len(), 3);
3458 assert_eq!(values.value(0), 1);
3459 assert_eq!(values.value(1), 2);
3460 assert_eq!(values.value(2), 3);
3461 }
3462
3463 fn encode_avro_long_extreme(value: i64) -> Vec<u8> {
3466 let mut n = ((value << 1) ^ (value >> 63)) as u64;
3467 let mut out = Vec::new();
3468 while n >= 0x80 {
3469 out.push((n as u8) | 0x80);
3470 n >>= 7;
3471 }
3472 out.push(n as u8);
3473 out
3474 }
3475
3476 fn array_of_null_decoder() -> Decoder {
3479 let list_dt = avro_from_codec(Codec::List(Arc::new(avro_from_codec(Codec::Null))));
3480 Decoder::try_new(&list_dt).unwrap()
3481 }
3482
3483 #[test]
3484 fn test_array_of_null_decodes() {
3485 let mut decoder = array_of_null_decoder();
3486 let mut data = encode_avro_long(3); data.extend_from_slice(&encode_avro_long(0)); decoder.decode(&mut AvroCursor::new(&data)).unwrap();
3489 }
3490
3491 #[test]
3492 fn test_array_block_count_i64_max_errors() {
3493 let mut decoder = array_of_null_decoder();
3495 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();
3498 assert!(
3499 err.to_string().contains("Capacity overflow"),
3500 "unexpected error: {err}",
3501 );
3502 }
3503
3504 #[test]
3505 fn test_array_block_count_i64_min_errors() {
3506 let mut decoder = array_of_null_decoder();
3508 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();
3511 assert!(
3512 err.to_string().contains("Capacity overflow"),
3513 "unexpected error: {err}",
3514 );
3515 }
3516
3517 #[test]
3518 fn test_nested_array_decoding() {
3519 let inner_ty = avro_from_codec(Codec::List(Arc::new(avro_from_codec(Codec::Int32))));
3520 let nested_ty = avro_from_codec(Codec::List(Arc::new(inner_ty.clone())));
3521 let mut decoder = Decoder::try_new(&nested_ty).unwrap();
3522 let mut buf = Vec::new();
3523 buf.extend(encode_avro_long(1));
3524 buf.extend(encode_avro_long(2));
3525 buf.extend(encode_avro_int(5));
3526 buf.extend(encode_avro_int(6));
3527 buf.extend(encode_avro_long(0));
3528 buf.extend(encode_avro_long(0));
3529 let mut cursor = AvroCursor::new(&buf);
3530 decoder.decode(&mut cursor).unwrap();
3531 let arr = decoder.flush(None).unwrap();
3532 let outer = arr.as_any().downcast_ref::<ListArray>().unwrap();
3533 assert_eq!(outer.len(), 1);
3534 assert_eq!(outer.value_length(0), 1);
3535 let inner = outer.values().as_any().downcast_ref::<ListArray>().unwrap();
3536 assert_eq!(inner.len(), 1);
3537 assert_eq!(inner.value_length(0), 2);
3538 let values = inner
3539 .values()
3540 .as_any()
3541 .downcast_ref::<Int32Array>()
3542 .unwrap();
3543 assert_eq!(values.values(), &[5, 6]);
3544 }
3545
3546 #[test]
3547 fn test_array_decoding_empty_array() {
3548 let value_type = avro_from_codec(Codec::Utf8);
3549 let map_type = avro_from_codec(Codec::List(Arc::new(value_type)));
3550 let mut decoder = Decoder::try_new(&map_type).unwrap();
3551 let data = encode_avro_long(0);
3552 decoder.decode(&mut AvroCursor::new(&data)).unwrap();
3553 let array = decoder.flush(None).unwrap();
3554 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3555 assert_eq!(list_arr.len(), 1);
3556 assert_eq!(list_arr.value_length(0), 0);
3557 }
3558
3559 #[test]
3560 fn test_array_decoding_writer_nonunion_items_reader_nullable_items() {
3561 use crate::schema::Array;
3562 let writer_schema = Schema::Complex(ComplexType::Array(Array {
3563 items: Box::new(Schema::TypeName(TypeName::Primitive(PrimitiveType::Int))),
3564 attributes: Attributes::default(),
3565 }));
3566 let reader_schema = Schema::Complex(ComplexType::Array(Array {
3567 items: Box::new(Schema::Union(vec![
3568 Schema::TypeName(TypeName::Primitive(PrimitiveType::Null)),
3569 Schema::TypeName(TypeName::Primitive(PrimitiveType::Int)),
3570 ])),
3571 attributes: Attributes::default(),
3572 }));
3573 let dt = resolved_root_datatype(writer_schema, reader_schema, false, false);
3574 if let Codec::List(inner) = dt.codec() {
3575 assert_eq!(
3576 inner.nullability(),
3577 Some(Nullability::NullFirst),
3578 "items should be nullable"
3579 );
3580 } else {
3581 panic!("expected List codec");
3582 }
3583 let mut decoder = Decoder::try_new(&dt).unwrap();
3584 let mut data = encode_avro_long(2);
3585 data.extend(encode_avro_int(10));
3586 data.extend(encode_avro_int(20));
3587 data.extend(encode_avro_long(0));
3588 let mut cursor = AvroCursor::new(&data);
3589 decoder.decode(&mut cursor).unwrap();
3590 assert_eq!(
3591 cursor.position(),
3592 data.len(),
3593 "all bytes should be consumed"
3594 );
3595 let array = decoder.flush(None).unwrap();
3596 let list_arr = array.as_any().downcast_ref::<ListArray>().unwrap();
3597 assert_eq!(list_arr.len(), 1, "one list/row");
3598 assert_eq!(list_arr.value_length(0), 2, "two items in the list");
3599 let values = list_arr.values().as_primitive::<Int32Type>();
3600 assert_eq!(values.len(), 2);
3601 assert_eq!(values.value(0), 10);
3602 assert_eq!(values.value(1), 20);
3603 assert!(!values.is_null(0));
3604 assert!(!values.is_null(1));
3605 }
3606
3607 #[test]
3608 fn test_decimal_decoding_fixed256() {
3609 let dt = avro_from_codec(Codec::Decimal(50, Some(2), Some(32)));
3610 let mut decoder = Decoder::try_new(&dt).unwrap();
3611 let row1 = [
3612 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3613 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3614 0x00, 0x00, 0x30, 0x39,
3615 ];
3616 let row2 = [
3617 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3618 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3619 0xFF, 0xFF, 0xFF, 0x85,
3620 ];
3621 let mut data = Vec::new();
3622 data.extend_from_slice(&row1);
3623 data.extend_from_slice(&row2);
3624 let mut cursor = AvroCursor::new(&data);
3625 decoder.decode(&mut cursor).unwrap();
3626 decoder.decode(&mut cursor).unwrap();
3627 let arr = decoder.flush(None).unwrap();
3628 let dec = arr.as_any().downcast_ref::<Decimal256Array>().unwrap();
3629 assert_eq!(dec.len(), 2);
3630 assert_eq!(dec.value_as_string(0), "123.45");
3631 assert_eq!(dec.value_as_string(1), "-1.23");
3632 }
3633
3634 #[test]
3635 fn test_decimal_decoding_fixed128() {
3636 let dt = avro_from_codec(Codec::Decimal(28, Some(2), Some(16)));
3637 let mut decoder = Decoder::try_new(&dt).unwrap();
3638 let row1 = [
3639 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3640 0x30, 0x39,
3641 ];
3642 let row2 = [
3643 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3644 0xFF, 0x85,
3645 ];
3646 let mut data = Vec::new();
3647 data.extend_from_slice(&row1);
3648 data.extend_from_slice(&row2);
3649 let mut cursor = AvroCursor::new(&data);
3650 decoder.decode(&mut cursor).unwrap();
3651 decoder.decode(&mut cursor).unwrap();
3652 let arr = decoder.flush(None).unwrap();
3653 let dec = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3654 assert_eq!(dec.len(), 2);
3655 assert_eq!(dec.value_as_string(0), "123.45");
3656 assert_eq!(dec.value_as_string(1), "-1.23");
3657 }
3658
3659 #[test]
3660 fn test_decimal_decoding_fixed32_from_32byte_fixed_storage() {
3661 let dt = avro_from_codec(Codec::Decimal(5, Some(2), Some(32)));
3662 let mut decoder = Decoder::try_new(&dt).unwrap();
3663 let row1 = [
3664 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3665 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3666 0x00, 0x00, 0x30, 0x39,
3667 ];
3668 let row2 = [
3669 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3670 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3671 0xFF, 0xFF, 0xFF, 0x85,
3672 ];
3673 let mut data = Vec::new();
3674 data.extend_from_slice(&row1);
3675 data.extend_from_slice(&row2);
3676 let mut cursor = AvroCursor::new(&data);
3677 decoder.decode(&mut cursor).unwrap();
3678 decoder.decode(&mut cursor).unwrap();
3679 let arr = decoder.flush(None).unwrap();
3680 #[cfg(feature = "small_decimals")]
3681 {
3682 let dec = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3683 assert_eq!(dec.len(), 2);
3684 assert_eq!(dec.value_as_string(0), "123.45");
3685 assert_eq!(dec.value_as_string(1), "-1.23");
3686 }
3687 #[cfg(not(feature = "small_decimals"))]
3688 {
3689 let dec = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3690 assert_eq!(dec.len(), 2);
3691 assert_eq!(dec.value_as_string(0), "123.45");
3692 assert_eq!(dec.value_as_string(1), "-1.23");
3693 }
3694 }
3695
3696 #[test]
3697 fn test_decimal_decoding_fixed32_from_16byte_fixed_storage() {
3698 let dt = avro_from_codec(Codec::Decimal(5, Some(2), Some(16)));
3699 let mut decoder = Decoder::try_new(&dt).unwrap();
3700 let row1 = [
3701 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
3702 0x30, 0x39,
3703 ];
3704 let row2 = [
3705 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
3706 0xFF, 0x85,
3707 ];
3708 let mut data = Vec::new();
3709 data.extend_from_slice(&row1);
3710 data.extend_from_slice(&row2);
3711 let mut cursor = AvroCursor::new(&data);
3712 decoder.decode(&mut cursor).unwrap();
3713 decoder.decode(&mut cursor).unwrap();
3714
3715 let arr = decoder.flush(None).unwrap();
3716 #[cfg(feature = "small_decimals")]
3717 {
3718 let dec = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3719 assert_eq!(dec.len(), 2);
3720 assert_eq!(dec.value_as_string(0), "123.45");
3721 assert_eq!(dec.value_as_string(1), "-1.23");
3722 }
3723 #[cfg(not(feature = "small_decimals"))]
3724 {
3725 let dec = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3726 assert_eq!(dec.len(), 2);
3727 assert_eq!(dec.value_as_string(0), "123.45");
3728 assert_eq!(dec.value_as_string(1), "-1.23");
3729 }
3730 }
3731
3732 #[test]
3733 fn test_decimal_decoding_bytes_with_nulls() {
3734 let dt = avro_from_codec(Codec::Decimal(4, Some(1), None));
3735 let inner = Decoder::try_new(&dt).unwrap();
3736 let mut decoder = Decoder::Nullable(
3737 NullablePlan::ReadTag {
3738 nullability: Nullability::NullSecond,
3739 resolution: ResolutionPlan::Promotion(Promotion::Direct),
3740 },
3741 NullBufferBuilder::new(DEFAULT_CAPACITY),
3742 Box::new(inner),
3743 );
3744 let mut data = Vec::new();
3745 data.extend_from_slice(&encode_avro_int(0));
3746 data.extend_from_slice(&encode_avro_bytes(&[0x04, 0xD2]));
3747 data.extend_from_slice(&encode_avro_int(1));
3748 data.extend_from_slice(&encode_avro_int(0));
3749 data.extend_from_slice(&encode_avro_bytes(&[0xFB, 0x2E]));
3750 let mut cursor = AvroCursor::new(&data);
3751 decoder.decode(&mut cursor).unwrap();
3752 decoder.decode(&mut cursor).unwrap();
3753 decoder.decode(&mut cursor).unwrap();
3754 let arr = decoder.flush(None).unwrap();
3755 #[cfg(feature = "small_decimals")]
3756 {
3757 let dec_arr = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3758 assert_eq!(dec_arr.len(), 3);
3759 assert!(dec_arr.is_valid(0));
3760 assert!(!dec_arr.is_valid(1));
3761 assert!(dec_arr.is_valid(2));
3762 assert_eq!(dec_arr.value_as_string(0), "123.4");
3763 assert_eq!(dec_arr.value_as_string(2), "-123.4");
3764 }
3765 #[cfg(not(feature = "small_decimals"))]
3766 {
3767 let dec_arr = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3768 assert_eq!(dec_arr.len(), 3);
3769 assert!(dec_arr.is_valid(0));
3770 assert!(!dec_arr.is_valid(1));
3771 assert!(dec_arr.is_valid(2));
3772 assert_eq!(dec_arr.value_as_string(0), "123.4");
3773 assert_eq!(dec_arr.value_as_string(2), "-123.4");
3774 }
3775 }
3776
3777 #[test]
3778 fn test_decimal_decoding_bytes_with_nulls_fixed_size_narrow_result() {
3779 let dt = avro_from_codec(Codec::Decimal(6, Some(2), Some(16)));
3780 let inner = Decoder::try_new(&dt).unwrap();
3781 let mut decoder = Decoder::Nullable(
3782 NullablePlan::ReadTag {
3783 nullability: Nullability::NullSecond,
3784 resolution: ResolutionPlan::Promotion(Promotion::Direct),
3785 },
3786 NullBufferBuilder::new(DEFAULT_CAPACITY),
3787 Box::new(inner),
3788 );
3789 let row1 = [
3790 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
3791 0xE2, 0x40,
3792 ];
3793 let row3 = [
3794 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFE,
3795 0x1D, 0xC0,
3796 ];
3797 let mut data = Vec::new();
3798 data.extend_from_slice(&encode_avro_int(0));
3799 data.extend_from_slice(&row1);
3800 data.extend_from_slice(&encode_avro_int(1));
3801 data.extend_from_slice(&encode_avro_int(0));
3802 data.extend_from_slice(&row3);
3803 let mut cursor = AvroCursor::new(&data);
3804 decoder.decode(&mut cursor).unwrap();
3805 decoder.decode(&mut cursor).unwrap();
3806 decoder.decode(&mut cursor).unwrap();
3807 let arr = decoder.flush(None).unwrap();
3808 #[cfg(feature = "small_decimals")]
3809 {
3810 let dec_arr = arr.as_any().downcast_ref::<Decimal32Array>().unwrap();
3811 assert_eq!(dec_arr.len(), 3);
3812 assert!(dec_arr.is_valid(0));
3813 assert!(!dec_arr.is_valid(1));
3814 assert!(dec_arr.is_valid(2));
3815 assert_eq!(dec_arr.value_as_string(0), "1234.56");
3816 assert_eq!(dec_arr.value_as_string(2), "-1234.56");
3817 }
3818 #[cfg(not(feature = "small_decimals"))]
3819 {
3820 let dec_arr = arr.as_any().downcast_ref::<Decimal128Array>().unwrap();
3821 assert_eq!(dec_arr.len(), 3);
3822 assert!(dec_arr.is_valid(0));
3823 assert!(!dec_arr.is_valid(1));
3824 assert!(dec_arr.is_valid(2));
3825 assert_eq!(dec_arr.value_as_string(0), "1234.56");
3826 assert_eq!(dec_arr.value_as_string(2), "-1234.56");
3827 }
3828 }
3829
3830 #[test]
3831 fn test_enum_decoding() {
3832 let symbols: Arc<[String]> = vec!["A", "B", "C"].into_iter().map(String::from).collect();
3833 let avro_type = avro_from_codec(Codec::Enum(symbols.clone()));
3834 let mut decoder = Decoder::try_new(&avro_type).unwrap();
3835 let mut data = Vec::new();
3836 data.extend_from_slice(&encode_avro_int(2));
3837 data.extend_from_slice(&encode_avro_int(0));
3838 data.extend_from_slice(&encode_avro_int(1));
3839 let mut cursor = AvroCursor::new(&data);
3840 decoder.decode(&mut cursor).unwrap();
3841 decoder.decode(&mut cursor).unwrap();
3842 decoder.decode(&mut cursor).unwrap();
3843 let array = decoder.flush(None).unwrap();
3844 let dict_array = array
3845 .as_any()
3846 .downcast_ref::<DictionaryArray<Int32Type>>()
3847 .unwrap();
3848 assert_eq!(dict_array.len(), 3);
3849 let values = dict_array
3850 .values()
3851 .as_any()
3852 .downcast_ref::<StringArray>()
3853 .unwrap();
3854 assert_eq!(values.value(0), "A");
3855 assert_eq!(values.value(1), "B");
3856 assert_eq!(values.value(2), "C");
3857 assert_eq!(dict_array.keys().values(), &[2, 0, 1]);
3858 }
3859
3860 #[test]
3861 fn test_enum_decoding_with_nulls() {
3862 let symbols: Arc<[String]> = vec!["X", "Y"].into_iter().map(String::from).collect();
3863 let enum_codec = Codec::Enum(symbols.clone());
3864 let avro_type =
3865 AvroDataType::new(enum_codec, Default::default(), Some(Nullability::NullFirst));
3866 let mut decoder = Decoder::try_new(&avro_type).unwrap();
3867 let mut data = Vec::new();
3868 data.extend_from_slice(&encode_avro_long(1));
3869 data.extend_from_slice(&encode_avro_int(1));
3870 data.extend_from_slice(&encode_avro_long(0));
3871 data.extend_from_slice(&encode_avro_long(1));
3872 data.extend_from_slice(&encode_avro_int(0));
3873 let mut cursor = AvroCursor::new(&data);
3874 decoder.decode(&mut cursor).unwrap();
3875 decoder.decode(&mut cursor).unwrap();
3876 decoder.decode(&mut cursor).unwrap();
3877 let array = decoder.flush(None).unwrap();
3878 let dict_array = array
3879 .as_any()
3880 .downcast_ref::<DictionaryArray<Int32Type>>()
3881 .unwrap();
3882 assert_eq!(dict_array.len(), 3);
3883 assert!(dict_array.is_valid(0));
3884 assert!(dict_array.is_null(1));
3885 assert!(dict_array.is_valid(2));
3886 let expected_keys = Int32Array::from(vec![Some(1), None, Some(0)]);
3887 assert_eq!(dict_array.keys(), &expected_keys);
3888 let values = dict_array
3889 .values()
3890 .as_any()
3891 .downcast_ref::<StringArray>()
3892 .unwrap();
3893 assert_eq!(values.value(0), "X");
3894 assert_eq!(values.value(1), "Y");
3895 }
3896
3897 #[test]
3898 fn test_duration_decoding_with_nulls() {
3899 let duration_codec = Codec::Interval;
3900 let avro_type = AvroDataType::new(
3901 duration_codec,
3902 Default::default(),
3903 Some(Nullability::NullFirst),
3904 );
3905 let mut decoder = Decoder::try_new(&avro_type).unwrap();
3906 let mut data = Vec::new();
3907 data.extend_from_slice(&encode_avro_long(1)); let mut duration1 = Vec::new();
3910 duration1.extend_from_slice(&1u32.to_le_bytes());
3911 duration1.extend_from_slice(&2u32.to_le_bytes());
3912 duration1.extend_from_slice(&3u32.to_le_bytes());
3913 data.extend_from_slice(&duration1);
3914 data.extend_from_slice(&encode_avro_long(0)); data.extend_from_slice(&encode_avro_long(1)); let mut duration2 = Vec::new();
3918 duration2.extend_from_slice(&4u32.to_le_bytes());
3919 duration2.extend_from_slice(&5u32.to_le_bytes());
3920 duration2.extend_from_slice(&6u32.to_le_bytes());
3921 data.extend_from_slice(&duration2);
3922 let mut cursor = AvroCursor::new(&data);
3923 decoder.decode(&mut cursor).unwrap();
3924 decoder.decode(&mut cursor).unwrap();
3925 decoder.decode(&mut cursor).unwrap();
3926 let array = decoder.flush(None).unwrap();
3927 let interval_array = array
3928 .as_any()
3929 .downcast_ref::<IntervalMonthDayNanoArray>()
3930 .unwrap();
3931 assert_eq!(interval_array.len(), 3);
3932 assert!(interval_array.is_valid(0));
3933 assert!(interval_array.is_null(1));
3934 assert!(interval_array.is_valid(2));
3935 let expected = IntervalMonthDayNanoArray::from(vec![
3936 Some(IntervalMonthDayNano {
3937 months: 1,
3938 days: 2,
3939 nanoseconds: 3_000_000,
3940 }),
3941 None,
3942 Some(IntervalMonthDayNano {
3943 months: 4,
3944 days: 5,
3945 nanoseconds: 6_000_000,
3946 }),
3947 ]);
3948 assert_eq!(interval_array, &expected);
3949 }
3950
3951 #[cfg(feature = "avro_custom_types")]
3952 #[test]
3953 fn test_interval_month_day_nano_custom_decoding_with_nulls() {
3954 let avro_type = AvroDataType::new(
3955 Codec::IntervalMonthDayNano,
3956 Default::default(),
3957 Some(Nullability::NullFirst),
3958 );
3959 let mut decoder = Decoder::try_new(&avro_type).unwrap();
3960 let mut data = Vec::new();
3961 data.extend_from_slice(&encode_avro_long(1));
3963 data.extend_from_slice(&1i32.to_le_bytes());
3964 data.extend_from_slice(&(-2i32).to_le_bytes());
3965 data.extend_from_slice(&3i64.to_le_bytes());
3966 data.extend_from_slice(&encode_avro_long(0));
3968 data.extend_from_slice(&encode_avro_long(1));
3970 data.extend_from_slice(&(-4i32).to_le_bytes());
3971 data.extend_from_slice(&5i32.to_le_bytes());
3972 data.extend_from_slice(&(-6i64).to_le_bytes());
3973 let mut cursor = AvroCursor::new(&data);
3974 decoder.decode(&mut cursor).unwrap();
3975 decoder.decode(&mut cursor).unwrap();
3976 decoder.decode(&mut cursor).unwrap();
3977 let array = decoder.flush(None).unwrap();
3978 let interval_array = array
3979 .as_any()
3980 .downcast_ref::<IntervalMonthDayNanoArray>()
3981 .unwrap();
3982 assert_eq!(interval_array.len(), 3);
3983 let expected = IntervalMonthDayNanoArray::from(vec![
3984 Some(IntervalMonthDayNano::new(1, -2, 3)),
3985 None,
3986 Some(IntervalMonthDayNano::new(-4, 5, -6)),
3987 ]);
3988 assert_eq!(interval_array, &expected);
3989 }
3990
3991 #[test]
3992 fn test_duration_decoding_empty() {
3993 let duration_codec = Codec::Interval;
3994 let avro_type = AvroDataType::new(duration_codec, Default::default(), None);
3995 let mut decoder = Decoder::try_new(&avro_type).unwrap();
3996 let array = decoder.flush(None).unwrap();
3997 assert_eq!(array.len(), 0);
3998 }
3999
4000 #[test]
4001 #[cfg(feature = "avro_custom_types")]
4002 fn test_duration_seconds_decoding() {
4003 let avro_type = AvroDataType::new(Codec::DurationSeconds, Default::default(), None);
4004 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4005 let mut data = Vec::new();
4006 data.extend_from_slice(&encode_avro_long(0));
4008 data.extend_from_slice(&encode_avro_long(-1));
4009 data.extend_from_slice(&encode_avro_long(2));
4010 let mut cursor = AvroCursor::new(&data);
4011 decoder.decode(&mut cursor).unwrap();
4012 decoder.decode(&mut cursor).unwrap();
4013 decoder.decode(&mut cursor).unwrap();
4014 let array = decoder.flush(None).unwrap();
4015 let dur = array
4016 .as_any()
4017 .downcast_ref::<DurationSecondArray>()
4018 .unwrap();
4019 assert_eq!(dur.values(), &[0, -1, 2]);
4020 }
4021
4022 #[test]
4023 #[cfg(feature = "avro_custom_types")]
4024 fn test_duration_milliseconds_decoding() {
4025 let avro_type = AvroDataType::new(Codec::DurationMillis, Default::default(), None);
4026 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4027 let mut data = Vec::new();
4028 for v in [1i64, 0, -2] {
4029 data.extend_from_slice(&encode_avro_long(v));
4030 }
4031 let mut cursor = AvroCursor::new(&data);
4032 for _ in 0..3 {
4033 decoder.decode(&mut cursor).unwrap();
4034 }
4035 let array = decoder.flush(None).unwrap();
4036 let dur = array
4037 .as_any()
4038 .downcast_ref::<DurationMillisecondArray>()
4039 .unwrap();
4040 assert_eq!(dur.values(), &[1, 0, -2]);
4041 }
4042
4043 #[test]
4044 #[cfg(feature = "avro_custom_types")]
4045 fn test_duration_microseconds_decoding() {
4046 let avro_type = AvroDataType::new(Codec::DurationMicros, Default::default(), None);
4047 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4048 let mut data = Vec::new();
4049 for v in [5i64, -6, 7] {
4050 data.extend_from_slice(&encode_avro_long(v));
4051 }
4052 let mut cursor = AvroCursor::new(&data);
4053 for _ in 0..3 {
4054 decoder.decode(&mut cursor).unwrap();
4055 }
4056 let array = decoder.flush(None).unwrap();
4057 let dur = array
4058 .as_any()
4059 .downcast_ref::<DurationMicrosecondArray>()
4060 .unwrap();
4061 assert_eq!(dur.values(), &[5, -6, 7]);
4062 }
4063
4064 #[test]
4065 #[cfg(feature = "avro_custom_types")]
4066 fn test_duration_nanoseconds_decoding() {
4067 let avro_type = AvroDataType::new(Codec::DurationNanos, Default::default(), None);
4068 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4069 let mut data = Vec::new();
4070 for v in [8i64, 9, -10] {
4071 data.extend_from_slice(&encode_avro_long(v));
4072 }
4073 let mut cursor = AvroCursor::new(&data);
4074 for _ in 0..3 {
4075 decoder.decode(&mut cursor).unwrap();
4076 }
4077 let array = decoder.flush(None).unwrap();
4078 let dur = array
4079 .as_any()
4080 .downcast_ref::<DurationNanosecondArray>()
4081 .unwrap();
4082 assert_eq!(dur.values(), &[8, 9, -10]);
4083 }
4084
4085 #[test]
4086 fn test_nullable_decode_error_bitmap_corruption() {
4087 let avro_type = AvroDataType::new(
4089 Codec::Int32,
4090 Default::default(),
4091 Some(Nullability::NullSecond),
4092 );
4093 let mut decoder = Decoder::try_new(&avro_type).unwrap();
4094
4095 let mut row1 = Vec::new();
4097 row1.extend_from_slice(&encode_avro_int(1));
4098
4099 let mut row2 = Vec::new();
4101 row2.extend_from_slice(&encode_avro_int(0)); let mut row3 = Vec::new();
4105 row3.extend_from_slice(&encode_avro_int(0)); row3.extend_from_slice(&encode_avro_int(42)); decoder.decode(&mut AvroCursor::new(&row1)).unwrap();
4109 assert!(decoder.decode(&mut AvroCursor::new(&row2)).is_err()); decoder.decode(&mut AvroCursor::new(&row3)).unwrap();
4111
4112 let array = decoder.flush(None).unwrap();
4113
4114 assert_eq!(array.len(), 2);
4116 let int_array = array.as_any().downcast_ref::<Int32Array>().unwrap();
4117 assert!(int_array.is_null(0)); assert_eq!(int_array.value(1), 42); }
4120
4121 #[test]
4122 fn test_enum_mapping_reordered_symbols() {
4123 let reader_symbols: Arc<[String]> =
4124 vec!["B".to_string(), "C".to_string(), "A".to_string()].into();
4125 let mapping: Arc<[i32]> = Arc::from(vec![2, 0, 1]);
4126 let default_index: i32 = -1;
4127 let mut dec = Decoder::Enum(
4128 Vec::with_capacity(DEFAULT_CAPACITY),
4129 reader_symbols.clone(),
4130 Some(EnumResolution {
4131 mapping,
4132 default_index,
4133 }),
4134 );
4135 let mut data = Vec::new();
4136 data.extend_from_slice(&encode_avro_int(0));
4137 data.extend_from_slice(&encode_avro_int(1));
4138 data.extend_from_slice(&encode_avro_int(2));
4139 let mut cur = AvroCursor::new(&data);
4140 dec.decode(&mut cur).unwrap();
4141 dec.decode(&mut cur).unwrap();
4142 dec.decode(&mut cur).unwrap();
4143 let arr = dec.flush(None).unwrap();
4144 let dict = arr
4145 .as_any()
4146 .downcast_ref::<DictionaryArray<Int32Type>>()
4147 .unwrap();
4148 let expected_keys = Int32Array::from(vec![2, 0, 1]);
4149 assert_eq!(dict.keys(), &expected_keys);
4150 let values = dict
4151 .values()
4152 .as_any()
4153 .downcast_ref::<StringArray>()
4154 .unwrap();
4155 assert_eq!(values.value(0), "B");
4156 assert_eq!(values.value(1), "C");
4157 assert_eq!(values.value(2), "A");
4158 }
4159
4160 #[test]
4161 fn test_enum_mapping_unknown_symbol_and_out_of_range_fall_back_to_default() {
4162 let reader_symbols: Arc<[String]> = vec!["A".to_string(), "B".to_string()].into();
4163 let default_index: i32 = 1;
4164 let mapping: Arc<[i32]> = Arc::from(vec![0, 1]);
4165 let mut dec = Decoder::Enum(
4166 Vec::with_capacity(DEFAULT_CAPACITY),
4167 reader_symbols.clone(),
4168 Some(EnumResolution {
4169 mapping,
4170 default_index,
4171 }),
4172 );
4173 let mut data = Vec::new();
4174 data.extend_from_slice(&encode_avro_int(0));
4175 data.extend_from_slice(&encode_avro_int(1));
4176 data.extend_from_slice(&encode_avro_int(99));
4177 let mut cur = AvroCursor::new(&data);
4178 dec.decode(&mut cur).unwrap();
4179 dec.decode(&mut cur).unwrap();
4180 dec.decode(&mut cur).unwrap();
4181 let arr = dec.flush(None).unwrap();
4182 let dict = arr
4183 .as_any()
4184 .downcast_ref::<DictionaryArray<Int32Type>>()
4185 .unwrap();
4186 let expected_keys = Int32Array::from(vec![0, 1, 1]);
4187 assert_eq!(dict.keys(), &expected_keys);
4188 let values = dict
4189 .values()
4190 .as_any()
4191 .downcast_ref::<StringArray>()
4192 .unwrap();
4193 assert_eq!(values.value(0), "A");
4194 assert_eq!(values.value(1), "B");
4195 }
4196
4197 #[test]
4198 fn test_enum_mapping_unknown_symbol_without_default_errors() {
4199 let reader_symbols: Arc<[String]> = vec!["A".to_string()].into();
4200 let default_index: i32 = -1; let mapping: Arc<[i32]> = Arc::from(vec![-1]);
4202 let mut dec = Decoder::Enum(
4203 Vec::with_capacity(DEFAULT_CAPACITY),
4204 reader_symbols,
4205 Some(EnumResolution {
4206 mapping,
4207 default_index,
4208 }),
4209 );
4210 let data = encode_avro_int(0);
4211 let mut cur = AvroCursor::new(&data);
4212 let err = dec
4213 .decode(&mut cur)
4214 .expect_err("expected decode error for unresolved enum without default");
4215 let msg = err.to_string();
4216 assert!(
4217 msg.contains("not resolvable") && msg.contains("no default"),
4218 "unexpected error message: {msg}"
4219 );
4220 }
4221
4222 fn make_record_resolved_decoder(
4223 reader_fields: &[(&str, DataType, bool)],
4224 writer_projections: Vec<FieldProjection>,
4225 ) -> Decoder {
4226 let mut field_refs: Vec<FieldRef> = Vec::with_capacity(reader_fields.len());
4227 let mut encodings: Vec<Decoder> = Vec::with_capacity(reader_fields.len());
4228 for (name, dt, nullable) in reader_fields {
4229 field_refs.push(Arc::new(ArrowField::new(*name, dt.clone(), *nullable)));
4230 let enc = match dt {
4231 DataType::Int32 => Decoder::Int32(Vec::new()),
4232 DataType::Int64 => Decoder::Int64(Vec::new()),
4233 DataType::Utf8 => {
4234 Decoder::String(OffsetBufferBuilder::new(DEFAULT_CAPACITY), Vec::new())
4235 }
4236 other => panic!("Unsupported test reader field type: {other:?}"),
4237 };
4238 encodings.push(enc);
4239 }
4240 let fields: Fields = field_refs.into();
4241 Decoder::Record(
4242 fields,
4243 encodings,
4244 vec![None; reader_fields.len()],
4245 Some(Projector {
4246 writer_projections,
4247 default_injections: Arc::from(Vec::<(usize, AvroLiteral)>::new()),
4248 }),
4249 )
4250 }
4251
4252 #[test]
4253 fn test_skip_writer_trailing_field_int32() {
4254 let mut dec = make_record_resolved_decoder(
4255 &[("id", arrow_schema::DataType::Int32, false)],
4256 vec![
4257 FieldProjection::ToReader(0),
4258 FieldProjection::Skip(super::Skipper::Int32),
4259 ],
4260 );
4261 let mut data = Vec::new();
4262 data.extend_from_slice(&encode_avro_int(7));
4263 data.extend_from_slice(&encode_avro_int(999));
4264 let mut cur = AvroCursor::new(&data);
4265 dec.decode(&mut cur).unwrap();
4266 assert_eq!(cur.position(), data.len());
4267 let arr = dec.flush(None).unwrap();
4268 let struct_arr = arr.as_any().downcast_ref::<StructArray>().unwrap();
4269 assert_eq!(struct_arr.len(), 1);
4270 let id = struct_arr
4271 .column_by_name("id")
4272 .unwrap()
4273 .as_any()
4274 .downcast_ref::<Int32Array>()
4275 .unwrap();
4276 assert_eq!(id.value(0), 7);
4277 }
4278
4279 #[test]
4280 fn test_skip_writer_middle_field_string() {
4281 let mut dec = make_record_resolved_decoder(
4282 &[
4283 ("id", DataType::Int32, false),
4284 ("score", DataType::Int64, false),
4285 ],
4286 vec![
4287 FieldProjection::ToReader(0),
4288 FieldProjection::Skip(Skipper::String),
4289 FieldProjection::ToReader(1),
4290 ],
4291 );
4292 let mut data = Vec::new();
4293 data.extend_from_slice(&encode_avro_int(42));
4294 data.extend_from_slice(&encode_avro_bytes(b"abcdef"));
4295 data.extend_from_slice(&encode_avro_long(1000));
4296 let mut cur = AvroCursor::new(&data);
4297 dec.decode(&mut cur).unwrap();
4298 assert_eq!(cur.position(), data.len());
4299 let arr = dec.flush(None).unwrap();
4300 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4301 let id = s
4302 .column_by_name("id")
4303 .unwrap()
4304 .as_any()
4305 .downcast_ref::<Int32Array>()
4306 .unwrap();
4307 let score = s
4308 .column_by_name("score")
4309 .unwrap()
4310 .as_any()
4311 .downcast_ref::<Int64Array>()
4312 .unwrap();
4313 assert_eq!(id.value(0), 42);
4314 assert_eq!(score.value(0), 1000);
4315 }
4316
4317 #[test]
4318 fn test_skip_writer_array_with_negative_block_count_fast() {
4319 let mut dec = make_record_resolved_decoder(
4320 &[("id", DataType::Int32, false)],
4321 vec![
4322 FieldProjection::Skip(super::Skipper::List(Box::new(Skipper::Int32))),
4323 FieldProjection::ToReader(0),
4324 ],
4325 );
4326 let mut array_payload = Vec::new();
4327 array_payload.extend_from_slice(&encode_avro_int(1));
4328 array_payload.extend_from_slice(&encode_avro_int(2));
4329 array_payload.extend_from_slice(&encode_avro_int(3));
4330 let mut data = Vec::new();
4331 data.extend_from_slice(&encode_avro_long(-3));
4332 data.extend_from_slice(&encode_avro_long(array_payload.len() as i64));
4333 data.extend_from_slice(&array_payload);
4334 data.extend_from_slice(&encode_avro_long(0));
4335 data.extend_from_slice(&encode_avro_int(5));
4336 let mut cur = AvroCursor::new(&data);
4337 dec.decode(&mut cur).unwrap();
4338 assert_eq!(cur.position(), data.len());
4339 let arr = dec.flush(None).unwrap();
4340 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4341 let id = s
4342 .column_by_name("id")
4343 .unwrap()
4344 .as_any()
4345 .downcast_ref::<Int32Array>()
4346 .unwrap();
4347 assert_eq!(id.len(), 1);
4348 assert_eq!(id.value(0), 5);
4349 }
4350
4351 #[test]
4352 fn test_skip_writer_map_with_negative_block_count_fast() {
4353 let mut dec = make_record_resolved_decoder(
4354 &[("id", DataType::Int32, false)],
4355 vec![
4356 FieldProjection::Skip(Skipper::Map(Box::new(Skipper::Int32))),
4357 FieldProjection::ToReader(0),
4358 ],
4359 );
4360 let mut entries = Vec::new();
4361 entries.extend_from_slice(&encode_avro_bytes(b"k1"));
4362 entries.extend_from_slice(&encode_avro_int(10));
4363 entries.extend_from_slice(&encode_avro_bytes(b"k2"));
4364 entries.extend_from_slice(&encode_avro_int(20));
4365 let mut data = Vec::new();
4366 data.extend_from_slice(&encode_avro_long(-2));
4367 data.extend_from_slice(&encode_avro_long(entries.len() as i64));
4368 data.extend_from_slice(&entries);
4369 data.extend_from_slice(&encode_avro_long(0));
4370 data.extend_from_slice(&encode_avro_int(123));
4371 let mut cur = AvroCursor::new(&data);
4372 dec.decode(&mut cur).unwrap();
4373 assert_eq!(cur.position(), data.len());
4374 let arr = dec.flush(None).unwrap();
4375 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4376 let id = s
4377 .column_by_name("id")
4378 .unwrap()
4379 .as_any()
4380 .downcast_ref::<Int32Array>()
4381 .unwrap();
4382 assert_eq!(id.len(), 1);
4383 assert_eq!(id.value(0), 123);
4384 }
4385
4386 #[test]
4387 fn test_skip_writer_nullable_field_union_nullfirst() {
4388 let mut dec = make_record_resolved_decoder(
4389 &[("id", DataType::Int32, false)],
4390 vec![
4391 FieldProjection::Skip(super::Skipper::Nullable(
4392 Nullability::NullFirst,
4393 Box::new(super::Skipper::Int32),
4394 )),
4395 FieldProjection::ToReader(0),
4396 ],
4397 );
4398 let mut row1 = Vec::new();
4399 row1.extend_from_slice(&encode_avro_long(0));
4400 row1.extend_from_slice(&encode_avro_int(5));
4401 let mut row2 = Vec::new();
4402 row2.extend_from_slice(&encode_avro_long(1));
4403 row2.extend_from_slice(&encode_avro_int(123));
4404 row2.extend_from_slice(&encode_avro_int(7));
4405 let mut cur1 = AvroCursor::new(&row1);
4406 let mut cur2 = AvroCursor::new(&row2);
4407 dec.decode(&mut cur1).unwrap();
4408 dec.decode(&mut cur2).unwrap();
4409 assert_eq!(cur1.position(), row1.len());
4410 assert_eq!(cur2.position(), row2.len());
4411 let arr = dec.flush(None).unwrap();
4412 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
4413 let id = s
4414 .column_by_name("id")
4415 .unwrap()
4416 .as_any()
4417 .downcast_ref::<Int32Array>()
4418 .unwrap();
4419 assert_eq!(id.len(), 2);
4420 assert_eq!(id.value(0), 5);
4421 assert_eq!(id.value(1), 7);
4422 }
4423
4424 fn make_dense_union_avro(
4425 children: Vec<(Codec, &'_ str, DataType)>,
4426 type_ids: Vec<i8>,
4427 ) -> AvroDataType {
4428 let mut avro_children: Vec<AvroDataType> = Vec::with_capacity(children.len());
4429 let mut fields: Vec<arrow_schema::Field> = Vec::with_capacity(children.len());
4430 for (codec, name, dt) in children {
4431 avro_children.push(AvroDataType::new(codec, Default::default(), None));
4432 fields.push(arrow_schema::Field::new(name, dt, true));
4433 }
4434 let union_fields = UnionFields::try_new(type_ids, fields).unwrap();
4435 let union_codec = Codec::Union(avro_children.into(), union_fields, UnionMode::Dense);
4436 AvroDataType::new(union_codec, Default::default(), None)
4437 }
4438
4439 #[test]
4440 fn test_union_dense_two_children_custom_type_ids() {
4441 let union_dt = make_dense_union_avro(
4442 vec![
4443 (Codec::Int32, "i", DataType::Int32),
4444 (Codec::Utf8, "s", DataType::Utf8),
4445 ],
4446 vec![2, 5],
4447 );
4448 let mut dec = Decoder::try_new(&union_dt).unwrap();
4449 let mut r1 = Vec::new();
4450 r1.extend_from_slice(&encode_avro_long(0));
4451 r1.extend_from_slice(&encode_avro_int(7));
4452 let mut r2 = Vec::new();
4453 r2.extend_from_slice(&encode_avro_long(1));
4454 r2.extend_from_slice(&encode_avro_bytes(b"x"));
4455 let mut r3 = Vec::new();
4456 r3.extend_from_slice(&encode_avro_long(0));
4457 r3.extend_from_slice(&encode_avro_int(-1));
4458 dec.decode(&mut AvroCursor::new(&r1)).unwrap();
4459 dec.decode(&mut AvroCursor::new(&r2)).unwrap();
4460 dec.decode(&mut AvroCursor::new(&r3)).unwrap();
4461 let array = dec.flush(None).unwrap();
4462 let ua = array
4463 .as_any()
4464 .downcast_ref::<UnionArray>()
4465 .expect("expected UnionArray");
4466 assert_eq!(ua.len(), 3);
4467 assert_eq!(ua.type_id(0), 2);
4468 assert_eq!(ua.type_id(1), 5);
4469 assert_eq!(ua.type_id(2), 2);
4470 assert_eq!(ua.value_offset(0), 0);
4471 assert_eq!(ua.value_offset(1), 0);
4472 assert_eq!(ua.value_offset(2), 1);
4473 let int_child = ua
4474 .child(2)
4475 .as_any()
4476 .downcast_ref::<Int32Array>()
4477 .expect("int child");
4478 assert_eq!(int_child.len(), 2);
4479 assert_eq!(int_child.value(0), 7);
4480 assert_eq!(int_child.value(1), -1);
4481 let str_child = ua
4482 .child(5)
4483 .as_any()
4484 .downcast_ref::<StringArray>()
4485 .expect("string child");
4486 assert_eq!(str_child.len(), 1);
4487 assert_eq!(str_child.value(0), "x");
4488 }
4489
4490 #[test]
4491 fn test_union_dense_with_null_and_string_children() {
4492 let union_dt = make_dense_union_avro(
4493 vec![
4494 (Codec::Null, "n", DataType::Null),
4495 (Codec::Utf8, "s", DataType::Utf8),
4496 ],
4497 vec![42, 7],
4498 );
4499 let mut dec = Decoder::try_new(&union_dt).unwrap();
4500 let r1 = encode_avro_long(0);
4501 let mut r2 = Vec::new();
4502 r2.extend_from_slice(&encode_avro_long(1));
4503 r2.extend_from_slice(&encode_avro_bytes(b"abc"));
4504 let r3 = encode_avro_long(0);
4505 dec.decode(&mut AvroCursor::new(&r1)).unwrap();
4506 dec.decode(&mut AvroCursor::new(&r2)).unwrap();
4507 dec.decode(&mut AvroCursor::new(&r3)).unwrap();
4508 let array = dec.flush(None).unwrap();
4509 let ua = array
4510 .as_any()
4511 .downcast_ref::<UnionArray>()
4512 .expect("expected UnionArray");
4513 assert_eq!(ua.len(), 3);
4514 assert_eq!(ua.type_id(0), 42);
4515 assert_eq!(ua.type_id(1), 7);
4516 assert_eq!(ua.type_id(2), 42);
4517 assert_eq!(ua.value_offset(0), 0);
4518 assert_eq!(ua.value_offset(1), 0);
4519 assert_eq!(ua.value_offset(2), 1);
4520 let null_child = ua
4521 .child(42)
4522 .as_any()
4523 .downcast_ref::<NullArray>()
4524 .expect("null child");
4525 assert_eq!(null_child.len(), 2);
4526 let str_child = ua
4527 .child(7)
4528 .as_any()
4529 .downcast_ref::<StringArray>()
4530 .expect("string child");
4531 assert_eq!(str_child.len(), 1);
4532 assert_eq!(str_child.value(0), "abc");
4533 }
4534
4535 #[test]
4536 fn test_union_decode_negative_branch_index_errors() {
4537 let union_dt = make_dense_union_avro(
4538 vec![
4539 (Codec::Int32, "i", DataType::Int32),
4540 (Codec::Utf8, "s", DataType::Utf8),
4541 ],
4542 vec![0, 1],
4543 );
4544 let mut dec = Decoder::try_new(&union_dt).unwrap();
4545 let row = encode_avro_long(-1); let err = dec
4547 .decode(&mut AvroCursor::new(&row))
4548 .expect_err("expected error for negative branch index");
4549 let msg = err.to_string();
4550 assert!(
4551 msg.contains("Negative union branch index"),
4552 "unexpected error message: {msg}"
4553 );
4554 }
4555
4556 #[test]
4557 fn test_union_decode_out_of_range_branch_index_errors() {
4558 let union_dt = make_dense_union_avro(
4559 vec![
4560 (Codec::Int32, "i", DataType::Int32),
4561 (Codec::Utf8, "s", DataType::Utf8),
4562 ],
4563 vec![10, 11],
4564 );
4565 let mut dec = Decoder::try_new(&union_dt).unwrap();
4566 let row = encode_avro_long(2);
4567 let err = dec
4568 .decode(&mut AvroCursor::new(&row))
4569 .expect_err("expected error for out-of-range branch index");
4570 let msg = err.to_string();
4571 assert!(
4572 msg.contains("out of range"),
4573 "unexpected error message: {msg}"
4574 );
4575 }
4576
4577 #[test]
4578 fn test_union_sparse_mode_not_supported() {
4579 let children: Vec<AvroDataType> = vec![
4580 AvroDataType::new(Codec::Int32, Default::default(), None),
4581 AvroDataType::new(Codec::Utf8, Default::default(), None),
4582 ];
4583 let uf = UnionFields::try_new(
4584 vec![1, 3],
4585 vec![
4586 arrow_schema::Field::new("i", DataType::Int32, true),
4587 arrow_schema::Field::new("s", DataType::Utf8, true),
4588 ],
4589 )
4590 .unwrap();
4591 let codec = Codec::Union(children.into(), uf, UnionMode::Sparse);
4592 let dt = AvroDataType::new(codec, Default::default(), None);
4593 let err = Decoder::try_new(&dt).expect_err("sparse union should not be supported");
4594 let msg = err.to_string();
4595 assert!(
4596 msg.contains("Sparse Arrow unions are not yet supported"),
4597 "unexpected error message: {msg}"
4598 );
4599 }
4600
4601 fn make_record_decoder_with_projector_defaults(
4602 reader_fields: &[(&str, DataType, bool)],
4603 field_defaults: Vec<Option<AvroLiteral>>,
4604 default_injections: Vec<(usize, AvroLiteral)>,
4605 ) -> Decoder {
4606 assert_eq!(
4607 field_defaults.len(),
4608 reader_fields.len(),
4609 "field_defaults must have one entry per reader field"
4610 );
4611 let mut field_refs: Vec<FieldRef> = Vec::with_capacity(reader_fields.len());
4612 let mut encodings: Vec<Decoder> = Vec::with_capacity(reader_fields.len());
4613 for (name, dt, nullable) in reader_fields {
4614 field_refs.push(Arc::new(ArrowField::new(*name, dt.clone(), *nullable)));
4615 let enc = match dt {
4616 DataType::Int32 => Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY)),
4617 DataType::Int64 => Decoder::Int64(Vec::with_capacity(DEFAULT_CAPACITY)),
4618 DataType::Utf8 => Decoder::String(
4619 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4620 Vec::with_capacity(DEFAULT_CAPACITY),
4621 ),
4622 other => panic!("Unsupported test field type in helper: {other:?}"),
4623 };
4624 encodings.push(enc);
4625 }
4626 let fields: Fields = field_refs.into();
4627 let projector = Projector {
4628 writer_projections: vec![],
4629 default_injections: Arc::from(default_injections),
4630 };
4631 Decoder::Record(fields, encodings, field_defaults, Some(projector))
4632 }
4633
4634 #[cfg(feature = "avro_custom_types")]
4635 #[test]
4636 fn test_default_append_custom_integer_range_validation() {
4637 let mut d_i8 = Decoder::Int8(Vec::with_capacity(DEFAULT_CAPACITY));
4638 d_i8.append_default(&AvroLiteral::Int(i8::MIN as i32))
4639 .unwrap();
4640 d_i8.append_default(&AvroLiteral::Int(i8::MAX as i32))
4641 .unwrap();
4642 let err_i8_high = d_i8
4643 .append_default(&AvroLiteral::Int(i8::MAX as i32 + 1))
4644 .unwrap_err();
4645 assert!(err_i8_high.to_string().contains("out of range for i8"));
4646 let err_i8_low = d_i8
4647 .append_default(&AvroLiteral::Int(i8::MIN as i32 - 1))
4648 .unwrap_err();
4649 assert!(err_i8_low.to_string().contains("out of range for i8"));
4650 let arr_i8 = d_i8.flush(None).unwrap();
4651 let values_i8 = arr_i8.as_any().downcast_ref::<Int8Array>().unwrap();
4652 assert_eq!(values_i8.values(), &[i8::MIN, i8::MAX]);
4653
4654 let mut d_i16 = Decoder::Int16(Vec::with_capacity(DEFAULT_CAPACITY));
4655 d_i16
4656 .append_default(&AvroLiteral::Int(i16::MIN as i32))
4657 .unwrap();
4658 d_i16
4659 .append_default(&AvroLiteral::Int(i16::MAX as i32))
4660 .unwrap();
4661 let err_i16_high = d_i16
4662 .append_default(&AvroLiteral::Int(i16::MAX as i32 + 1))
4663 .unwrap_err();
4664 assert!(err_i16_high.to_string().contains("out of range for i16"));
4665 let err_i16_low = d_i16
4666 .append_default(&AvroLiteral::Int(i16::MIN as i32 - 1))
4667 .unwrap_err();
4668 assert!(err_i16_low.to_string().contains("out of range for i16"));
4669 let arr_i16 = d_i16.flush(None).unwrap();
4670 let values_i16 = arr_i16.as_any().downcast_ref::<Int16Array>().unwrap();
4671 assert_eq!(values_i16.values(), &[i16::MIN, i16::MAX]);
4672
4673 let mut d_u8 = Decoder::UInt8(Vec::with_capacity(DEFAULT_CAPACITY));
4674 d_u8.append_default(&AvroLiteral::Int(0)).unwrap();
4675 d_u8.append_default(&AvroLiteral::Int(u8::MAX as i32))
4676 .unwrap();
4677 let err_u8_neg = d_u8.append_default(&AvroLiteral::Int(-1)).unwrap_err();
4678 assert!(err_u8_neg.to_string().contains("out of range for u8"));
4679 let err_u8_high = d_u8
4680 .append_default(&AvroLiteral::Int(u8::MAX as i32 + 1))
4681 .unwrap_err();
4682 assert!(err_u8_high.to_string().contains("out of range for u8"));
4683 let arr_u8 = d_u8.flush(None).unwrap();
4684 let values_u8 = arr_u8.as_any().downcast_ref::<UInt8Array>().unwrap();
4685 assert_eq!(values_u8.values(), &[0, u8::MAX]);
4686
4687 let mut d_u16 = Decoder::UInt16(Vec::with_capacity(DEFAULT_CAPACITY));
4688 d_u16.append_default(&AvroLiteral::Int(0)).unwrap();
4689 d_u16
4690 .append_default(&AvroLiteral::Int(u16::MAX as i32))
4691 .unwrap();
4692 let err_u16_neg = d_u16.append_default(&AvroLiteral::Int(-1)).unwrap_err();
4693 assert!(err_u16_neg.to_string().contains("out of range for u16"));
4694 let err_u16_high = d_u16
4695 .append_default(&AvroLiteral::Int(u16::MAX as i32 + 1))
4696 .unwrap_err();
4697 assert!(err_u16_high.to_string().contains("out of range for u16"));
4698 let arr_u16 = d_u16.flush(None).unwrap();
4699 let values_u16 = arr_u16.as_any().downcast_ref::<UInt16Array>().unwrap();
4700 assert_eq!(values_u16.values(), &[0, u16::MAX]);
4701
4702 let mut d_u32 = Decoder::UInt32(Vec::with_capacity(DEFAULT_CAPACITY));
4703 d_u32.append_default(&AvroLiteral::Long(0)).unwrap();
4704 d_u32
4705 .append_default(&AvroLiteral::Long(u32::MAX as i64))
4706 .unwrap();
4707 let err_u32_neg = d_u32.append_default(&AvroLiteral::Long(-1)).unwrap_err();
4708 assert!(err_u32_neg.to_string().contains("out of range for u32"));
4709 let err_u32_high = d_u32
4710 .append_default(&AvroLiteral::Long(u32::MAX as i64 + 1))
4711 .unwrap_err();
4712 assert!(err_u32_high.to_string().contains("out of range for u32"));
4713 let arr_u32 = d_u32.flush(None).unwrap();
4714 let values_u32 = arr_u32.as_any().downcast_ref::<UInt32Array>().unwrap();
4715 assert_eq!(values_u32.values(), &[0, u32::MAX]);
4716 }
4717
4718 #[cfg(feature = "avro_custom_types")]
4719 #[test]
4720 fn test_decode_custom_integer_range_validation() {
4721 let mut d_i8 = Decoder::try_new(&avro_from_codec(Codec::Int8)).unwrap();
4722 d_i8.decode(&mut AvroCursor::new(&encode_avro_int(i8::MIN as i32)))
4723 .unwrap();
4724 d_i8.decode(&mut AvroCursor::new(&encode_avro_int(i8::MAX as i32)))
4725 .unwrap();
4726 let err_i8_high = d_i8
4727 .decode(&mut AvroCursor::new(&encode_avro_int(i8::MAX as i32 + 1)))
4728 .unwrap_err();
4729 assert!(err_i8_high.to_string().contains("out of range for i8"));
4730 let err_i8_low = d_i8
4731 .decode(&mut AvroCursor::new(&encode_avro_int(i8::MIN as i32 - 1)))
4732 .unwrap_err();
4733 assert!(err_i8_low.to_string().contains("out of range for i8"));
4734 let arr_i8 = d_i8.flush(None).unwrap();
4735 let values_i8 = arr_i8.as_any().downcast_ref::<Int8Array>().unwrap();
4736 assert_eq!(values_i8.values(), &[i8::MIN, i8::MAX]);
4737
4738 let mut d_i16 = Decoder::try_new(&avro_from_codec(Codec::Int16)).unwrap();
4739 d_i16
4740 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MIN as i32)))
4741 .unwrap();
4742 d_i16
4743 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MAX as i32)))
4744 .unwrap();
4745 let err_i16_high = d_i16
4746 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MAX as i32 + 1)))
4747 .unwrap_err();
4748 assert!(err_i16_high.to_string().contains("out of range for i16"));
4749 let err_i16_low = d_i16
4750 .decode(&mut AvroCursor::new(&encode_avro_int(i16::MIN as i32 - 1)))
4751 .unwrap_err();
4752 assert!(err_i16_low.to_string().contains("out of range for i16"));
4753 let arr_i16 = d_i16.flush(None).unwrap();
4754 let values_i16 = arr_i16.as_any().downcast_ref::<Int16Array>().unwrap();
4755 assert_eq!(values_i16.values(), &[i16::MIN, i16::MAX]);
4756
4757 let mut d_u8 = Decoder::try_new(&avro_from_codec(Codec::UInt8)).unwrap();
4758 d_u8.decode(&mut AvroCursor::new(&encode_avro_int(0)))
4759 .unwrap();
4760 d_u8.decode(&mut AvroCursor::new(&encode_avro_int(u8::MAX as i32)))
4761 .unwrap();
4762 let err_u8_neg = d_u8
4763 .decode(&mut AvroCursor::new(&encode_avro_int(-1)))
4764 .unwrap_err();
4765 assert!(err_u8_neg.to_string().contains("out of range for u8"));
4766 let err_u8_high = d_u8
4767 .decode(&mut AvroCursor::new(&encode_avro_int(u8::MAX as i32 + 1)))
4768 .unwrap_err();
4769 assert!(err_u8_high.to_string().contains("out of range for u8"));
4770 let arr_u8 = d_u8.flush(None).unwrap();
4771 let values_u8 = arr_u8.as_any().downcast_ref::<UInt8Array>().unwrap();
4772 assert_eq!(values_u8.values(), &[0, u8::MAX]);
4773
4774 let mut d_u16 = Decoder::try_new(&avro_from_codec(Codec::UInt16)).unwrap();
4775 d_u16
4776 .decode(&mut AvroCursor::new(&encode_avro_int(0)))
4777 .unwrap();
4778 d_u16
4779 .decode(&mut AvroCursor::new(&encode_avro_int(u16::MAX as i32)))
4780 .unwrap();
4781 let err_u16_neg = d_u16
4782 .decode(&mut AvroCursor::new(&encode_avro_int(-1)))
4783 .unwrap_err();
4784 assert!(err_u16_neg.to_string().contains("out of range for u16"));
4785 let err_u16_high = d_u16
4786 .decode(&mut AvroCursor::new(&encode_avro_int(u16::MAX as i32 + 1)))
4787 .unwrap_err();
4788 assert!(err_u16_high.to_string().contains("out of range for u16"));
4789 let arr_u16 = d_u16.flush(None).unwrap();
4790 let values_u16 = arr_u16.as_any().downcast_ref::<UInt16Array>().unwrap();
4791 assert_eq!(values_u16.values(), &[0, u16::MAX]);
4792
4793 let mut d_u32 = Decoder::try_new(&avro_from_codec(Codec::UInt32)).unwrap();
4794 d_u32
4795 .decode(&mut AvroCursor::new(&encode_avro_long(0)))
4796 .unwrap();
4797 d_u32
4798 .decode(&mut AvroCursor::new(&encode_avro_long(u32::MAX as i64)))
4799 .unwrap();
4800 let err_u32_neg = d_u32
4801 .decode(&mut AvroCursor::new(&encode_avro_long(-1)))
4802 .unwrap_err();
4803 assert!(err_u32_neg.to_string().contains("out of range for u32"));
4804 let err_u32_high = d_u32
4805 .decode(&mut AvroCursor::new(&encode_avro_long(u32::MAX as i64 + 1)))
4806 .unwrap_err();
4807 assert!(err_u32_high.to_string().contains("out of range for u32"));
4808 let arr_u32 = d_u32.flush(None).unwrap();
4809 let values_u32 = arr_u32.as_any().downcast_ref::<UInt32Array>().unwrap();
4810 assert_eq!(values_u32.values(), &[0, u32::MAX]);
4811 }
4812
4813 #[test]
4814 fn test_default_append_int32_and_int64_from_int_and_long() {
4815 let mut d_i32 = Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY));
4816 d_i32.append_default(&AvroLiteral::Int(42)).unwrap();
4817 let arr = d_i32.flush(None).unwrap();
4818 let a = arr.as_any().downcast_ref::<Int32Array>().unwrap();
4819 assert_eq!(a.len(), 1);
4820 assert_eq!(a.value(0), 42);
4821 let mut d_i64 = Decoder::Int64(Vec::with_capacity(DEFAULT_CAPACITY));
4822 d_i64.append_default(&AvroLiteral::Int(5)).unwrap();
4823 d_i64.append_default(&AvroLiteral::Long(7)).unwrap();
4824 let arr64 = d_i64.flush(None).unwrap();
4825 let a64 = arr64.as_any().downcast_ref::<Int64Array>().unwrap();
4826 assert_eq!(a64.len(), 2);
4827 assert_eq!(a64.value(0), 5);
4828 assert_eq!(a64.value(1), 7);
4829 }
4830
4831 #[test]
4832 fn test_default_append_floats_and_doubles() {
4833 let mut d_f32 = Decoder::Float32(Vec::with_capacity(DEFAULT_CAPACITY));
4834 d_f32.append_default(&AvroLiteral::Float(1.5)).unwrap();
4835 let arr32 = d_f32.flush(None).unwrap();
4836 let a = arr32.as_any().downcast_ref::<Float32Array>().unwrap();
4837 assert_eq!(a.value(0), 1.5);
4838 let mut d_f64 = Decoder::Float64(Vec::with_capacity(DEFAULT_CAPACITY));
4839 d_f64.append_default(&AvroLiteral::Double(2.25)).unwrap();
4840 let arr64 = d_f64.flush(None).unwrap();
4841 let b = arr64.as_any().downcast_ref::<Float64Array>().unwrap();
4842 assert_eq!(b.value(0), 2.25);
4843 }
4844
4845 #[test]
4846 fn test_default_append_string_and_bytes() {
4847 let mut d_str = Decoder::String(
4848 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4849 Vec::with_capacity(DEFAULT_CAPACITY),
4850 );
4851 d_str
4852 .append_default(&AvroLiteral::String("hi".into()))
4853 .unwrap();
4854 let s_arr = d_str.flush(None).unwrap();
4855 let arr = s_arr.as_any().downcast_ref::<StringArray>().unwrap();
4856 assert_eq!(arr.value(0), "hi");
4857 let mut d_bytes = Decoder::Binary(
4858 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4859 Vec::with_capacity(DEFAULT_CAPACITY),
4860 );
4861 d_bytes
4862 .append_default(&AvroLiteral::Bytes(vec![1, 2, 3]))
4863 .unwrap();
4864 let b_arr = d_bytes.flush(None).unwrap();
4865 let barr = b_arr.as_any().downcast_ref::<BinaryArray>().unwrap();
4866 assert_eq!(barr.value(0), &[1, 2, 3]);
4867 let mut d_str_err = Decoder::String(
4868 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
4869 Vec::with_capacity(DEFAULT_CAPACITY),
4870 );
4871 let err = d_str_err
4872 .append_default(&AvroLiteral::Bytes(vec![0x61, 0x62]))
4873 .unwrap_err();
4874 assert!(
4875 err.to_string()
4876 .contains("Default for string must be string"),
4877 "unexpected error: {err:?}"
4878 );
4879 }
4880
4881 #[test]
4882 fn test_default_append_nullable_int32_null_and_value() {
4883 let inner = Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY));
4884 let mut dec = Decoder::Nullable(
4885 NullablePlan::ReadTag {
4886 nullability: Nullability::NullFirst,
4887 resolution: ResolutionPlan::Promotion(Promotion::Direct),
4888 },
4889 NullBufferBuilder::new(DEFAULT_CAPACITY),
4890 Box::new(inner),
4891 );
4892 dec.append_default(&AvroLiteral::Null).unwrap();
4893 dec.append_default(&AvroLiteral::Int(11)).unwrap();
4894 let arr = dec.flush(None).unwrap();
4895 let a = arr.as_any().downcast_ref::<Int32Array>().unwrap();
4896 assert_eq!(a.len(), 2);
4897 assert!(a.is_null(0));
4898 assert_eq!(a.value(1), 11);
4899 }
4900
4901 #[test]
4902 fn test_default_append_array_of_ints() {
4903 let list_dt = avro_from_codec(Codec::List(Arc::new(avro_from_codec(Codec::Int32))));
4904 let mut d = Decoder::try_new(&list_dt).unwrap();
4905 let items = vec![
4906 AvroLiteral::Int(1),
4907 AvroLiteral::Int(2),
4908 AvroLiteral::Int(3),
4909 ];
4910 d.append_default(&AvroLiteral::Array(items)).unwrap();
4911 let arr = d.flush(None).unwrap();
4912 let list = arr.as_any().downcast_ref::<ListArray>().unwrap();
4913 assert_eq!(list.len(), 1);
4914 assert_eq!(list.value_length(0), 3);
4915 let vals = list.values().as_any().downcast_ref::<Int32Array>().unwrap();
4916 assert_eq!(vals.values(), &[1, 2, 3]);
4917 }
4918
4919 #[test]
4920 fn test_default_append_map_string_to_int() {
4921 let map_dt = avro_from_codec(Codec::Map(Arc::new(avro_from_codec(Codec::Int32))));
4922 let mut d = Decoder::try_new(&map_dt).unwrap();
4923 let mut m: IndexMap<String, AvroLiteral> = IndexMap::new();
4924 m.insert("k1".to_string(), AvroLiteral::Int(10));
4925 m.insert("k2".to_string(), AvroLiteral::Int(20));
4926 d.append_default(&AvroLiteral::Map(m)).unwrap();
4927 let arr = d.flush(None).unwrap();
4928 let map = arr.as_any().downcast_ref::<MapArray>().unwrap();
4929 assert_eq!(map.len(), 1);
4930 assert_eq!(map.value_length(0), 2);
4931 let binding = map.value(0);
4932 let entries = binding.as_any().downcast_ref::<StructArray>().unwrap();
4933 let k = entries
4934 .column_by_name("key")
4935 .unwrap()
4936 .as_any()
4937 .downcast_ref::<StringArray>()
4938 .unwrap();
4939 let v = entries
4940 .column_by_name("value")
4941 .unwrap()
4942 .as_any()
4943 .downcast_ref::<Int32Array>()
4944 .unwrap();
4945 let keys: std::collections::HashSet<&str> = (0..k.len()).map(|i| k.value(i)).collect();
4946 assert_eq!(keys, ["k1", "k2"].into_iter().collect());
4947 let vals: std::collections::HashSet<i32> = (0..v.len()).map(|i| v.value(i)).collect();
4948 assert_eq!(vals, [10, 20].into_iter().collect());
4949 }
4950
4951 #[test]
4952 fn test_default_append_enum_by_symbol() {
4953 let symbols: Arc<[String]> = vec!["A".into(), "B".into(), "C".into()].into();
4954 let mut d = Decoder::Enum(Vec::with_capacity(DEFAULT_CAPACITY), symbols.clone(), None);
4955 d.append_default(&AvroLiteral::Enum("B".into())).unwrap();
4956 let arr = d.flush(None).unwrap();
4957 let dict = arr
4958 .as_any()
4959 .downcast_ref::<DictionaryArray<Int32Type>>()
4960 .unwrap();
4961 assert_eq!(dict.len(), 1);
4962 let expected = Int32Array::from(vec![1]);
4963 assert_eq!(dict.keys(), &expected);
4964 let values = dict
4965 .values()
4966 .as_any()
4967 .downcast_ref::<StringArray>()
4968 .unwrap();
4969 assert_eq!(values.value(1), "B");
4970 }
4971
4972 #[test]
4973 fn test_default_append_uuid_and_type_error() {
4974 let mut d = Decoder::Uuid(Vec::with_capacity(DEFAULT_CAPACITY));
4975 let uuid_str = "123e4567-e89b-12d3-a456-426614174000";
4976 d.append_default(&AvroLiteral::String(uuid_str.into()))
4977 .unwrap();
4978 let arr_ref = d.flush(None).unwrap();
4979 let arr = arr_ref
4980 .as_any()
4981 .downcast_ref::<FixedSizeBinaryArray>()
4982 .unwrap();
4983 assert_eq!(arr.value_length(), 16);
4984 assert_eq!(arr.len(), 1);
4985 let mut d2 = Decoder::Uuid(Vec::with_capacity(DEFAULT_CAPACITY));
4986 let err = d2
4987 .append_default(&AvroLiteral::Bytes(vec![0u8; 16]))
4988 .unwrap_err();
4989 assert!(
4990 err.to_string().contains("Default for uuid must be string"),
4991 "unexpected error: {err:?}"
4992 );
4993 }
4994
4995 #[test]
4996 fn test_default_append_fixed_and_length_mismatch() {
4997 let mut d = Decoder::Fixed(4, Vec::with_capacity(DEFAULT_CAPACITY));
4998 d.append_default(&AvroLiteral::Bytes(vec![1, 2, 3, 4]))
4999 .unwrap();
5000 let arr_ref = d.flush(None).unwrap();
5001 let arr = arr_ref
5002 .as_any()
5003 .downcast_ref::<FixedSizeBinaryArray>()
5004 .unwrap();
5005 assert_eq!(arr.value_length(), 4);
5006 assert_eq!(arr.value(0), &[1, 2, 3, 4]);
5007 let mut d_err = Decoder::Fixed(4, Vec::with_capacity(DEFAULT_CAPACITY));
5008 let err = d_err
5009 .append_default(&AvroLiteral::Bytes(vec![1, 2, 3]))
5010 .unwrap_err();
5011 assert!(
5012 err.to_string().contains("Fixed default length"),
5013 "unexpected error: {err:?}"
5014 );
5015 }
5016
5017 #[test]
5018 fn test_default_append_duration_and_length_validation() {
5019 let dt = avro_from_codec(Codec::Interval);
5020 let mut d = Decoder::try_new(&dt).unwrap();
5021 let mut bytes = Vec::with_capacity(12);
5022 bytes.extend_from_slice(&1u32.to_le_bytes());
5023 bytes.extend_from_slice(&2u32.to_le_bytes());
5024 bytes.extend_from_slice(&3u32.to_le_bytes());
5025 d.append_default(&AvroLiteral::Bytes(bytes)).unwrap();
5026 let arr_ref = d.flush(None).unwrap();
5027 let arr = arr_ref
5028 .as_any()
5029 .downcast_ref::<IntervalMonthDayNanoArray>()
5030 .unwrap();
5031 assert_eq!(arr.len(), 1);
5032 let v = arr.value(0);
5033 assert_eq!(v.months, 1);
5034 assert_eq!(v.days, 2);
5035 assert_eq!(v.nanoseconds, 3_000_000);
5036 let mut d_err = Decoder::try_new(&avro_from_codec(Codec::Interval)).unwrap();
5037 let err = d_err
5038 .append_default(&AvroLiteral::Bytes(vec![0u8; 11]))
5039 .unwrap_err();
5040 assert!(
5041 err.to_string()
5042 .contains("Duration default must be exactly 12 bytes"),
5043 "unexpected error: {err:?}"
5044 );
5045 }
5046
5047 #[test]
5048 fn test_default_append_decimal256_from_bytes() {
5049 let dt = avro_from_codec(Codec::Decimal(50, Some(2), Some(32)));
5050 let mut d = Decoder::try_new(&dt).unwrap();
5051 let pos: [u8; 32] = [
5052 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
5053 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
5054 0x00, 0x00, 0x30, 0x39,
5055 ];
5056 d.append_default(&AvroLiteral::Bytes(pos.to_vec())).unwrap();
5057 let neg: [u8; 32] = [
5058 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
5059 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
5060 0xFF, 0xFF, 0xFF, 0x85,
5061 ];
5062 d.append_default(&AvroLiteral::Bytes(neg.to_vec())).unwrap();
5063 let arr = d.flush(None).unwrap();
5064 let dec = arr.as_any().downcast_ref::<Decimal256Array>().unwrap();
5065 assert_eq!(dec.len(), 2);
5066 assert_eq!(dec.value_as_string(0), "123.45");
5067 assert_eq!(dec.value_as_string(1), "-1.23");
5068 }
5069
5070 #[test]
5071 fn test_record_append_default_map_missing_fields_uses_projector_field_defaults() {
5072 let field_defaults = vec![None, Some(AvroLiteral::String("hi".into()))];
5073 let mut rec = make_record_decoder_with_projector_defaults(
5074 &[("a", DataType::Int32, false), ("b", DataType::Utf8, false)],
5075 field_defaults,
5076 vec![],
5077 );
5078 let mut map: IndexMap<String, AvroLiteral> = IndexMap::new();
5079 map.insert("a".to_string(), AvroLiteral::Int(7));
5080 rec.append_default(&AvroLiteral::Map(map)).unwrap();
5081 let arr = rec.flush(None).unwrap();
5082 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5083 let a = s
5084 .column_by_name("a")
5085 .unwrap()
5086 .as_any()
5087 .downcast_ref::<Int32Array>()
5088 .unwrap();
5089 let b = s
5090 .column_by_name("b")
5091 .unwrap()
5092 .as_any()
5093 .downcast_ref::<StringArray>()
5094 .unwrap();
5095 assert_eq!(a.value(0), 7);
5096 assert_eq!(b.value(0), "hi");
5097 }
5098
5099 #[test]
5100 fn test_record_append_default_null_uses_projector_field_defaults() {
5101 let field_defaults = vec![
5102 Some(AvroLiteral::Int(5)),
5103 Some(AvroLiteral::String("x".into())),
5104 ];
5105 let mut rec = make_record_decoder_with_projector_defaults(
5106 &[("a", DataType::Int32, false), ("b", DataType::Utf8, false)],
5107 field_defaults,
5108 vec![],
5109 );
5110 rec.append_default(&AvroLiteral::Null).unwrap();
5111 let arr = rec.flush(None).unwrap();
5112 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5113 let a = s
5114 .column_by_name("a")
5115 .unwrap()
5116 .as_any()
5117 .downcast_ref::<Int32Array>()
5118 .unwrap();
5119 let b = s
5120 .column_by_name("b")
5121 .unwrap()
5122 .as_any()
5123 .downcast_ref::<StringArray>()
5124 .unwrap();
5125 assert_eq!(a.value(0), 5);
5126 assert_eq!(b.value(0), "x");
5127 }
5128
5129 #[test]
5130 fn test_record_append_default_missing_fields_without_projector_defaults_yields_type_nulls_or_empties()
5131 {
5132 let fields = vec![("a", DataType::Int32, true), ("b", DataType::Utf8, true)];
5133 let mut field_refs: Vec<FieldRef> = Vec::new();
5134 let mut encoders: Vec<Decoder> = Vec::new();
5135 for (name, dt, nullable) in &fields {
5136 field_refs.push(Arc::new(ArrowField::new(*name, dt.clone(), *nullable)));
5137 }
5138 let enc_a = Decoder::Nullable(
5139 NullablePlan::ReadTag {
5140 nullability: Nullability::NullSecond,
5141 resolution: ResolutionPlan::Promotion(Promotion::Direct),
5142 },
5143 NullBufferBuilder::new(DEFAULT_CAPACITY),
5144 Box::new(Decoder::Int32(Vec::with_capacity(DEFAULT_CAPACITY))),
5145 );
5146 let enc_b = Decoder::Nullable(
5147 NullablePlan::ReadTag {
5148 nullability: Nullability::NullSecond,
5149 resolution: ResolutionPlan::Promotion(Promotion::Direct),
5150 },
5151 NullBufferBuilder::new(DEFAULT_CAPACITY),
5152 Box::new(Decoder::String(
5153 OffsetBufferBuilder::new(DEFAULT_CAPACITY),
5154 Vec::with_capacity(DEFAULT_CAPACITY),
5155 )),
5156 );
5157 encoders.push(enc_a);
5158 encoders.push(enc_b);
5159 let field_defaults = vec![None, None]; let projector = Projector {
5161 writer_projections: vec![],
5162 default_injections: Arc::from(Vec::<(usize, AvroLiteral)>::new()),
5163 };
5164 let mut rec = Decoder::Record(field_refs.into(), encoders, field_defaults, Some(projector));
5165 let mut map: IndexMap<String, AvroLiteral> = IndexMap::new();
5166 map.insert("a".to_string(), AvroLiteral::Int(9));
5167 rec.append_default(&AvroLiteral::Map(map)).unwrap();
5168 let arr = rec.flush(None).unwrap();
5169 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5170 let a = s
5171 .column_by_name("a")
5172 .unwrap()
5173 .as_any()
5174 .downcast_ref::<Int32Array>()
5175 .unwrap();
5176 let b = s
5177 .column_by_name("b")
5178 .unwrap()
5179 .as_any()
5180 .downcast_ref::<StringArray>()
5181 .unwrap();
5182 assert!(a.is_valid(0));
5183 assert_eq!(a.value(0), 9);
5184 assert!(b.is_null(0));
5185 }
5186
5187 #[test]
5188 fn test_projector_default_injection_when_writer_lacks_fields() {
5189 let defaults = vec![None, None];
5190 let injections = vec![
5191 (0, AvroLiteral::Int(99)),
5192 (1, AvroLiteral::String("alice".into())),
5193 ];
5194 let mut rec = make_record_decoder_with_projector_defaults(
5195 &[
5196 ("id", DataType::Int32, false),
5197 ("name", DataType::Utf8, false),
5198 ],
5199 defaults,
5200 injections,
5201 );
5202 rec.decode(&mut AvroCursor::new(&[])).unwrap();
5203 let arr = rec.flush(None).unwrap();
5204 let s = arr.as_any().downcast_ref::<StructArray>().unwrap();
5205 let id = s
5206 .column_by_name("id")
5207 .unwrap()
5208 .as_any()
5209 .downcast_ref::<Int32Array>()
5210 .unwrap();
5211 let name = s
5212 .column_by_name("name")
5213 .unwrap()
5214 .as_any()
5215 .downcast_ref::<StringArray>()
5216 .unwrap();
5217 assert_eq!(id.value(0), 99);
5218 assert_eq!(name.value(0), "alice");
5219 }
5220
5221 #[test]
5222 fn union_type_ids_are_not_child_indexes() {
5223 let encodings: Vec<AvroDataType> =
5224 vec![avro_from_codec(Codec::Int32), avro_from_codec(Codec::Utf8)];
5225 let fields: UnionFields = [
5226 (42_i8, Arc::new(ArrowField::new("a", DataType::Int32, true))),
5227 (7_i8, Arc::new(ArrowField::new("b", DataType::Utf8, true))),
5228 ]
5229 .into_iter()
5230 .collect();
5231 let dt = avro_from_codec(Codec::Union(
5232 encodings.into(),
5233 fields.clone(),
5234 UnionMode::Dense,
5235 ));
5236 let mut dec = Decoder::try_new(&dt).expect("decoder");
5237 let mut b1 = encode_avro_long(1);
5238 b1.extend(encode_avro_bytes(b"hi"));
5239 dec.decode(&mut AvroCursor::new(&b1)).expect("decode b1");
5240 let mut b0 = encode_avro_long(0);
5241 b0.extend(encode_avro_int(5));
5242 dec.decode(&mut AvroCursor::new(&b0)).expect("decode b0");
5243 let arr = dec.flush(None).expect("flush");
5244 let ua = arr.as_any().downcast_ref::<UnionArray>().expect("union");
5245 assert_eq!(ua.len(), 2);
5246 assert_eq!(ua.type_id(0), 7, "type id must come from UnionFields");
5247 assert_eq!(ua.type_id(1), 42, "type id must come from UnionFields");
5248 assert_eq!(ua.value_offset(0), 0);
5249 assert_eq!(ua.value_offset(1), 0);
5250 let utf8_child = ua.child(7).as_any().downcast_ref::<StringArray>().unwrap();
5251 assert_eq!(utf8_child.len(), 1);
5252 assert_eq!(utf8_child.value(0), "hi");
5253 let int_child = ua.child(42).as_any().downcast_ref::<Int32Array>().unwrap();
5254 assert_eq!(int_child.len(), 1);
5255 assert_eq!(int_child.value(0), 5);
5256 let type_ids: Vec<i8> = fields.iter().map(|(tid, _)| tid).collect();
5257 assert_eq!(type_ids, vec![42_i8, 7_i8]);
5258 }
5259
5260 #[cfg(feature = "avro_custom_types")]
5261 #[test]
5262 fn skipper_from_avro_maps_custom_duration_variants_to_int64() -> Result<(), AvroError> {
5263 for codec in [
5264 Codec::DurationNanos,
5265 Codec::DurationMicros,
5266 Codec::DurationMillis,
5267 Codec::DurationSeconds,
5268 ] {
5269 let dt = make_avro_dt(codec.clone(), None);
5270 let s = Skipper::from_avro(&dt)?;
5271 match s {
5272 Skipper::Int64 => {}
5273 other => panic!("expected Int64 skipper for {codec:?}, got {other:?}"),
5274 }
5275 }
5276 Ok(())
5277 }
5278
5279 #[cfg(feature = "avro_custom_types")]
5280 #[test]
5281 fn skipper_skip_consumes_one_long_for_custom_durations() -> Result<(), AvroError> {
5282 let values: [i64; 7] = [0, 1, -1, 150, -150, i64::MAX / 3, i64::MIN / 3];
5283 for codec in [
5284 Codec::DurationNanos,
5285 Codec::DurationMicros,
5286 Codec::DurationMillis,
5287 Codec::DurationSeconds,
5288 ] {
5289 let dt = make_avro_dt(codec.clone(), None);
5290 let s = Skipper::from_avro(&dt)?;
5291 for &v in &values {
5292 let bytes = encode_avro_long(v);
5293 let mut cursor = AvroCursor::new(&bytes);
5294 s.skip(&mut cursor)?;
5295 assert_eq!(
5296 cursor.position(),
5297 bytes.len(),
5298 "did not consume all bytes for {codec:?} value {v}"
5299 );
5300 }
5301 }
5302 Ok(())
5303 }
5304
5305 #[cfg(feature = "avro_custom_types")]
5306 #[test]
5307 fn skipper_nullable_custom_duration_respects_null_first() -> Result<(), AvroError> {
5308 let dt = make_avro_dt(Codec::DurationNanos, Some(Nullability::NullFirst));
5309 let s = Skipper::from_avro(&dt)?;
5310 match &s {
5311 Skipper::Nullable(Nullability::NullFirst, inner) => match **inner {
5312 Skipper::Int64 => {}
5313 ref other => panic!("expected inner Int64, got {other:?}"),
5314 },
5315 other => panic!("expected Nullable(NullFirst, Int64), got {other:?}"),
5316 }
5317 {
5318 let buf = encode_vlq_u64(0);
5319 let mut cursor = AvroCursor::new(&buf);
5320 s.skip(&mut cursor)?;
5321 assert_eq!(cursor.position(), 1, "expected to consume only tag=0");
5322 }
5323 {
5324 let mut buf = encode_vlq_u64(1);
5325 buf.extend(encode_avro_long(0));
5326 let mut cursor = AvroCursor::new(&buf);
5327 s.skip(&mut cursor)?;
5328 assert_eq!(cursor.position(), 2, "expected to consume tag=1 + long(0)");
5329 }
5330
5331 Ok(())
5332 }
5333
5334 #[cfg(feature = "avro_custom_types")]
5335 #[test]
5336 fn skipper_nullable_custom_duration_respects_null_second() -> Result<(), AvroError> {
5337 let dt = make_avro_dt(Codec::DurationMicros, Some(Nullability::NullSecond));
5338 let s = Skipper::from_avro(&dt)?;
5339 match &s {
5340 Skipper::Nullable(Nullability::NullSecond, inner) => match **inner {
5341 Skipper::Int64 => {}
5342 ref other => panic!("expected inner Int64, got {other:?}"),
5343 },
5344 other => panic!("expected Nullable(NullSecond, Int64), got {other:?}"),
5345 }
5346 {
5347 let buf = encode_vlq_u64(1);
5348 let mut cursor = AvroCursor::new(&buf);
5349 s.skip(&mut cursor)?;
5350 assert_eq!(cursor.position(), 1, "expected to consume only tag=1");
5351 }
5352 {
5353 let mut buf = encode_vlq_u64(0);
5354 buf.extend(encode_avro_long(-1));
5355 let mut cursor = AvroCursor::new(&buf);
5356 s.skip(&mut cursor)?;
5357 assert_eq!(
5358 cursor.position(),
5359 1 + encode_avro_long(-1).len(),
5360 "expected to consume tag=0 + long(-1)"
5361 );
5362 }
5363 Ok(())
5364 }
5365
5366 #[test]
5367 fn skipper_interval_is_fixed12_and_skips_12_bytes() -> Result<(), AvroError> {
5368 let dt = make_avro_dt(Codec::Interval, None);
5369 let s = Skipper::from_avro(&dt)?;
5370 match s {
5371 Skipper::DurationFixed12 => {}
5372 other => panic!("expected DurationFixed12, got {other:?}"),
5373 }
5374 let payload = vec![0u8; 12];
5375 let mut cursor = AvroCursor::new(&payload);
5376 s.skip(&mut cursor)?;
5377 assert_eq!(cursor.position(), 12, "expected to consume 12 fixed bytes");
5378 Ok(())
5379 }
5380
5381 #[cfg(feature = "avro_custom_types")]
5382 #[test]
5383 fn test_run_end_encoded_width16_int32_basic_grouping() {
5384 use arrow_array::RunArray;
5385 use std::sync::Arc;
5386 let inner = avro_from_codec(Codec::Int32);
5387 let ree = AvroDataType::new(
5388 Codec::RunEndEncoded(Arc::new(inner), 16),
5389 Default::default(),
5390 None,
5391 );
5392 let mut dec = Decoder::try_new(&ree).expect("create REE decoder");
5393 for v in [1, 1, 1, 2, 2, 3, 3, 3, 3] {
5394 let bytes = encode_avro_int(v);
5395 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5396 }
5397 let arr = dec.flush(None).expect("flush");
5398 let ra = arr
5399 .as_any()
5400 .downcast_ref::<RunArray<Int16Type>>()
5401 .expect("RunArray<Int16Type>");
5402 assert_eq!(ra.len(), 9);
5403 assert_eq!(ra.run_ends().values(), &[3, 5, 9]);
5404 let vals = ra
5405 .values()
5406 .as_ref()
5407 .as_any()
5408 .downcast_ref::<Int32Array>()
5409 .expect("values Int32");
5410 assert_eq!(vals.values(), &[1, 2, 3]);
5411 }
5412
5413 #[cfg(feature = "avro_custom_types")]
5414 #[test]
5415 fn test_run_end_encoded_width32_nullable_values_group_nulls() {
5416 use arrow_array::RunArray;
5417 use std::sync::Arc;
5418 let inner = AvroDataType::new(
5419 Codec::Int32,
5420 Default::default(),
5421 Some(Nullability::NullSecond),
5422 );
5423 let ree = AvroDataType::new(
5424 Codec::RunEndEncoded(Arc::new(inner), 32),
5425 Default::default(),
5426 None,
5427 );
5428 let mut dec = Decoder::try_new(&ree).expect("create REE decoder");
5429 let seq: [Option<i32>; 8] = [
5430 None,
5431 None,
5432 Some(7),
5433 Some(7),
5434 Some(7),
5435 None,
5436 Some(5),
5437 Some(5),
5438 ];
5439 for item in seq {
5440 let mut bytes = Vec::new();
5441 match item {
5442 None => bytes.extend_from_slice(&encode_vlq_u64(1)),
5443 Some(v) => {
5444 bytes.extend_from_slice(&encode_vlq_u64(0));
5445 bytes.extend_from_slice(&encode_avro_int(v));
5446 }
5447 }
5448 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5449 }
5450 let arr = dec.flush(None).expect("flush");
5451 let ra = arr
5452 .as_any()
5453 .downcast_ref::<RunArray<Int32Type>>()
5454 .expect("RunArray<Int32Type>");
5455 assert_eq!(ra.len(), 8);
5456 assert_eq!(ra.run_ends().values(), &[2, 5, 6, 8]);
5457 let vals = ra
5458 .values()
5459 .as_ref()
5460 .as_any()
5461 .downcast_ref::<Int32Array>()
5462 .expect("values Int32 (nullable)");
5463 assert_eq!(vals.len(), 4);
5464 assert!(vals.is_null(0));
5465 assert_eq!(vals.value(1), 7);
5466 assert!(vals.is_null(2));
5467 assert_eq!(vals.value(3), 5);
5468 }
5469
5470 #[cfg(feature = "avro_custom_types")]
5471 #[test]
5472 fn test_run_end_encoded_decode_with_promotion_int_to_double_via_nullable_from_single() {
5473 use arrow_array::RunArray;
5474 let inner_values = Decoder::Float64(Vec::with_capacity(DEFAULT_CAPACITY));
5475 let ree = Decoder::RunEndEncoded(
5476 8, 0,
5478 Box::new(inner_values),
5479 );
5480 let mut dec = Decoder::Nullable(
5481 NullablePlan::FromSingle {
5482 resolution: ResolutionPlan::Promotion(Promotion::IntToDouble),
5483 },
5484 NullBufferBuilder::new(DEFAULT_CAPACITY),
5485 Box::new(ree),
5486 );
5487 for v in [1, 1, 2, 2, 2] {
5488 let bytes = encode_avro_int(v);
5489 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5490 }
5491 let arr = dec.flush(None).expect("flush");
5492 let ra = arr
5493 .as_any()
5494 .downcast_ref::<RunArray<Int64Type>>()
5495 .expect("RunArray<Int64Type>");
5496 assert_eq!(ra.len(), 5);
5497 assert_eq!(ra.run_ends().values(), &[2, 5]);
5498 let vals = ra
5499 .values()
5500 .as_ref()
5501 .as_any()
5502 .downcast_ref::<Float64Array>()
5503 .expect("values Float64");
5504 assert_eq!(vals.values(), &[1.0, 2.0]);
5505 }
5506
5507 #[cfg(feature = "avro_custom_types")]
5508 #[test]
5509 fn test_run_end_encoded_unsupported_run_end_width_errors() {
5510 use std::sync::Arc;
5511 let inner = avro_from_codec(Codec::Int32);
5512 let dt = AvroDataType::new(
5513 Codec::RunEndEncoded(Arc::new(inner), 3),
5514 Default::default(),
5515 None,
5516 );
5517 let err = Decoder::try_new(&dt).expect_err("must reject unsupported width");
5518 let msg = err.to_string();
5519 assert!(
5520 msg.contains("Unsupported run-end width")
5521 && msg.contains("16/32/64 bits or 2/4/8 bytes"),
5522 "unexpected error message: {msg}"
5523 );
5524 }
5525
5526 #[cfg(feature = "avro_custom_types")]
5527 #[test]
5528 fn test_run_end_encoded_empty_input_is_empty_runarray() {
5529 use arrow_array::RunArray;
5530 use std::sync::Arc;
5531 let inner = avro_from_codec(Codec::Utf8);
5532 let dt = AvroDataType::new(
5533 Codec::RunEndEncoded(Arc::new(inner), 4),
5534 Default::default(),
5535 None,
5536 );
5537 let mut dec = Decoder::try_new(&dt).expect("create REE decoder");
5538 let arr = dec.flush(None).expect("flush");
5539 let ra = arr
5540 .as_any()
5541 .downcast_ref::<RunArray<Int32Type>>()
5542 .expect("RunArray<Int32Type>");
5543 assert_eq!(ra.len(), 0);
5544 assert_eq!(ra.run_ends().len(), 0);
5545 assert_eq!(ra.values().len(), 0);
5546 }
5547
5548 #[cfg(feature = "avro_custom_types")]
5549 #[test]
5550 fn test_run_end_encoded_strings_grouping_width32_bits() {
5551 use arrow_array::RunArray;
5552 use std::sync::Arc;
5553 let inner = avro_from_codec(Codec::Utf8);
5554 let dt = AvroDataType::new(
5555 Codec::RunEndEncoded(Arc::new(inner), 32),
5556 Default::default(),
5557 None,
5558 );
5559 let mut dec = Decoder::try_new(&dt).expect("create REE decoder");
5560 for s in ["a", "a", "bb", "bb", "bb", "a"] {
5561 let bytes = encode_avro_bytes(s.as_bytes());
5562 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5563 }
5564 let arr = dec.flush(None).expect("flush");
5565 let ra = arr
5566 .as_any()
5567 .downcast_ref::<RunArray<Int32Type>>()
5568 .expect("RunArray<Int32Type>");
5569 assert_eq!(ra.run_ends().values(), &[2, 5, 6]);
5570 let vals = ra
5571 .values()
5572 .as_ref()
5573 .as_any()
5574 .downcast_ref::<StringArray>()
5575 .expect("values String");
5576 assert_eq!(vals.len(), 3);
5577 assert_eq!(vals.value(0), "a");
5578 assert_eq!(vals.value(1), "bb");
5579 assert_eq!(vals.value(2), "a");
5580 }
5581
5582 #[cfg(not(feature = "avro_custom_types"))]
5583 #[test]
5584 fn test_no_custom_types_feature_smoke_decodes_plain_int32() {
5585 let dt = avro_from_codec(Codec::Int32);
5586 let mut dec = Decoder::try_new(&dt).expect("create Int32 decoder");
5587 for v in [1, 2, 3] {
5588 let bytes = encode_avro_int(v);
5589 dec.decode(&mut AvroCursor::new(&bytes)).expect("decode");
5590 }
5591 let arr = dec.flush(None).expect("flush");
5592 let a = arr
5593 .as_any()
5594 .downcast_ref::<Int32Array>()
5595 .expect("Int32Array");
5596 assert_eq!(a.values(), &[1, 2, 3]);
5597 }
5598
5599 #[test]
5600 fn test_timestamp_nanos_decoding_offset_zero() {
5601 let avro_type = avro_from_codec(Codec::TimestampNanos(Some(Tz::OffsetZero)));
5602 let mut decoder = Decoder::try_new(&avro_type).expect("create TimestampNanos decoder");
5603 let mut data = Vec::new();
5604 for v in [0_i64, 1_i64, -1_i64, 1_234_567_890_i64] {
5605 data.extend_from_slice(&encode_avro_long(v));
5606 }
5607 let mut cur = AvroCursor::new(&data);
5608 for _ in 0..4 {
5609 decoder.decode(&mut cur).expect("decode nanos ts");
5610 }
5611 let array = decoder.flush(None).expect("flush nanos ts");
5612 let ts = array
5613 .as_any()
5614 .downcast_ref::<TimestampNanosecondArray>()
5615 .expect("TimestampNanosecondArray");
5616 assert_eq!(ts.values(), &[0, 1, -1, 1_234_567_890]);
5617 match ts.data_type() {
5618 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5619 assert_eq!(tz.as_deref(), Some("+00:00"));
5620 }
5621 other => panic!("expected Timestamp(Nanosecond, Some(\"+00:00\")), got {other:?}"),
5622 }
5623 }
5624
5625 #[test]
5626 fn test_timestamp_nanos_decoding_utc() {
5627 let avro_type = avro_from_codec(Codec::TimestampNanos(Some(Tz::Utc)));
5628 let mut decoder = Decoder::try_new(&avro_type).expect("create TimestampNanos decoder");
5629 let mut data = Vec::new();
5630 for v in [0_i64, 1_i64, -1_i64, 1_234_567_890_i64] {
5631 data.extend_from_slice(&encode_avro_long(v));
5632 }
5633 let mut cur = AvroCursor::new(&data);
5634 for _ in 0..4 {
5635 decoder.decode(&mut cur).expect("decode nanos ts");
5636 }
5637 let array = decoder.flush(None).expect("flush nanos ts");
5638 let ts = array
5639 .as_any()
5640 .downcast_ref::<TimestampNanosecondArray>()
5641 .expect("TimestampNanosecondArray");
5642 assert_eq!(ts.values(), &[0, 1, -1, 1_234_567_890]);
5643 match ts.data_type() {
5644 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5645 assert_eq!(tz.as_deref(), Some("UTC"));
5646 }
5647 other => panic!("expected Timestamp(Nanosecond, Some(\"UTC\")), got {other:?}"),
5648 }
5649 }
5650
5651 #[test]
5652 fn test_timestamp_nanos_decoding_local() {
5653 let avro_type = avro_from_codec(Codec::TimestampNanos(None));
5654 let mut decoder = Decoder::try_new(&avro_type).expect("create TimestampNanos decoder");
5655 let mut data = Vec::new();
5656 for v in [10_i64, 20_i64, -30_i64] {
5657 data.extend_from_slice(&encode_avro_long(v));
5658 }
5659 let mut cur = AvroCursor::new(&data);
5660 for _ in 0..3 {
5661 decoder.decode(&mut cur).expect("decode nanos ts");
5662 }
5663 let array = decoder.flush(None).expect("flush nanos ts");
5664 let ts = array
5665 .as_any()
5666 .downcast_ref::<TimestampNanosecondArray>()
5667 .expect("TimestampNanosecondArray");
5668 assert_eq!(ts.values(), &[10, 20, -30]);
5669 match ts.data_type() {
5670 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5671 assert_eq!(tz.as_deref(), None);
5672 }
5673 other => panic!("expected Timestamp(Nanosecond, None), got {other:?}"),
5674 }
5675 }
5676
5677 #[test]
5678 fn test_timestamp_nanos_decoding_with_nulls() {
5679 let avro_type = AvroDataType::new(
5680 Codec::TimestampNanos(None),
5681 Default::default(),
5682 Some(Nullability::NullFirst),
5683 );
5684 let mut decoder = Decoder::try_new(&avro_type).expect("create nullable TimestampNanos");
5685 let mut data = Vec::new();
5686 data.extend_from_slice(&encode_avro_long(1));
5687 data.extend_from_slice(&encode_avro_long(42));
5688 data.extend_from_slice(&encode_avro_long(0));
5689 data.extend_from_slice(&encode_avro_long(1));
5690 data.extend_from_slice(&encode_avro_long(-7));
5691 let mut cur = AvroCursor::new(&data);
5692 for _ in 0..3 {
5693 decoder.decode(&mut cur).expect("decode nullable nanos ts");
5694 }
5695 let array = decoder.flush(None).expect("flush nullable nanos ts");
5696 let ts = array
5697 .as_any()
5698 .downcast_ref::<TimestampNanosecondArray>()
5699 .expect("TimestampNanosecondArray");
5700 assert_eq!(ts.len(), 3);
5701 assert!(ts.is_valid(0));
5702 assert!(ts.is_null(1));
5703 assert!(ts.is_valid(2));
5704 assert_eq!(ts.value(0), 42);
5705 assert_eq!(ts.value(2), -7);
5706 match ts.data_type() {
5707 DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, tz) => {
5708 assert_eq!(tz.as_deref(), None);
5709 }
5710 other => panic!("expected Timestamp(Nanosecond, None), got {other:?}"),
5711 }
5712 }
5713}