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