mz_avro/
reader.rs

1// Copyright 2018 Flavien Raynaud.
2// Copyright Materialize, Inc. and contributors. All rights reserved.
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License in the LICENSE file at the
7// root of this repository, or online at
8//
9//     http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing, software
12// distributed under the License is distributed on an "AS IS" BASIS,
13// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14// See the License for the specific language governing permissions and
15// limitations under the License.
16//
17// This file is derived from the avro-rs project, available at
18// https://github.com/flavray/avro-rs. It was incorporated
19// directly into Materialize on March 3, 2020.
20//
21// The original source code is subject to the terms of the MIT license, a copy
22// of which can be found in the LICENSE file at the root of this repository.
23
24//! Logic handling reading from Avro format at user level.
25
26use std::collections::BTreeMap;
27use std::str::{FromStr, from_utf8};
28
29use aws_lc_rs::digest;
30use serde_json::from_slice;
31
32use crate::decode::{AvroRead, bound_block_object_count, decode};
33use crate::error::{DecodeError, Error as AvroError};
34use crate::schema::{
35    FullName, NamedSchemaPiece, ParseSchemaError, RecordField, ResolvedDefaultValueField,
36    ResolvedRecordField, Schema, SchemaNodeOrNamed, SchemaPiece, SchemaPieceOrNamed,
37    SchemaPieceRefOrNamed, resolve_schemas,
38};
39use crate::types::Value;
40use crate::{Codec, SchemaResolutionError, util};
41
42#[derive(Debug, Clone)]
43pub(crate) struct Header {
44    writer_schema: Schema,
45    marker: [u8; 16],
46    codec: Codec,
47}
48
49impl Header {
50    pub fn from_reader<R: AvroRead>(reader: &mut R) -> Result<Header, AvroError> {
51        let meta_schema = Schema {
52            named: vec![],
53            indices: Default::default(),
54            top: SchemaPiece::Map(Box::new(SchemaPiece::Bytes.into())).into(),
55        };
56
57        let mut buf = [0u8; 4];
58        reader.read_exact(&mut buf)?;
59
60        if buf != [b'O', b'b', b'j', 1u8] {
61            return Err(AvroError::Decode(DecodeError::WrongHeaderMagic(buf)));
62        }
63
64        if let Value::Map(meta) = decode(meta_schema.top_node(), reader)? {
65            // TODO: surface original parse schema errors instead of coalescing them here
66            let json = meta
67                .get("avro.schema")
68                .ok_or(AvroError::Decode(DecodeError::MissingAvroDotSchema))
69                .and_then(|bytes| {
70                    if let Value::Bytes(ref bytes) = *bytes {
71                        from_slice(bytes.as_ref()).map_err(|e| {
72                            AvroError::ParseSchema(ParseSchemaError::new(format!(
73                                "unable to decode schema bytes: {}",
74                                e
75                            )))
76                        })
77                    } else {
78                        unreachable!()
79                    }
80                })?;
81            let writer_schema = Schema::parse(&json).map_err(|e| {
82                ParseSchemaError::new(format!("unable to parse json as avro schema: {}", e))
83            })?;
84
85            let codec = meta
86                .get("avro.codec")
87                .map(|val| match val {
88                    Value::Bytes(bytes) => from_utf8(bytes.as_ref())
89                        .map_err(|_e| AvroError::Decode(DecodeError::CodecUtf8Error))
90                        .and_then(|codec| {
91                            Codec::from_str(codec).map_err(|_| {
92                                AvroError::Decode(DecodeError::UnrecognizedCodec(codec.to_string()))
93                            })
94                        }),
95                    _ => unreachable!(),
96                })
97                .unwrap_or(Ok(Codec::Null))?;
98
99            let mut marker = [0u8; 16];
100            reader.read_exact(&mut marker)?;
101
102            Ok(Header {
103                writer_schema,
104                marker,
105                codec,
106            })
107        } else {
108            unreachable!()
109        }
110    }
111
112    pub fn into_parts(self) -> (Schema, [u8; 16], Codec) {
113        (self.writer_schema, self.marker, self.codec)
114    }
115}
116
117pub struct Reader<R> {
118    header: Header,
119    inner: R,
120    errored: bool,
121    resolved_schema: Option<Schema>,
122    messages_remaining: usize,
123    // Internal buffering to reduce allocation.
124    buf: Vec<u8>,
125    buf_idx: usize,
126}
127
128/// An iterator over the `Block`s of a `Reader`
129pub struct BlockIter<R> {
130    inner: Reader<R>,
131}
132
133/// A block of Avro objects from an OCF file
134#[derive(Debug, Clone)]
135pub struct Block {
136    /// The raw bytes for the block
137    pub bytes: Vec<u8>,
138    /// The number of Avro objects in the block
139    pub len: usize,
140}
141
142impl<R: AvroRead> BlockIter<R> {
143    pub fn with_schema(reader_schema: &Schema, inner: R) -> Result<Self, AvroError> {
144        Ok(Self {
145            inner: Reader::with_schema(reader_schema, inner)?,
146        })
147    }
148}
149
150impl<R: AvroRead> Iterator for BlockIter<R> {
151    type Item = Result<Block, AvroError>;
152
153    fn next(&mut self) -> Option<Self::Item> {
154        assert!(self.inner.is_empty());
155
156        match self.inner.read_block_next() {
157            Ok(()) => {
158                if self.inner.is_empty() {
159                    None
160                } else {
161                    let bytes = std::mem::take(&mut self.inner.buf);
162                    let len = std::mem::take(&mut self.inner.messages_remaining);
163                    Some(Ok(Block { bytes, len }))
164                }
165            }
166            Err(e) => Some(Err(e)),
167        }
168    }
169}
170
171impl<R: AvroRead> Reader<R> {
172    /// Creates a `Reader` given something implementing the `tokio::io::AsyncRead` trait to read from.
173    /// No reader `Schema` will be set.
174    ///
175    /// **NOTE** The avro header is going to be read automatically upon creation of the `Reader`.
176    pub fn new(mut inner: R) -> Result<Reader<R>, AvroError> {
177        let header = Header::from_reader(&mut inner)?;
178        let reader = Reader {
179            header,
180            inner,
181            errored: false,
182            resolved_schema: None,
183            messages_remaining: 0,
184            buf: vec![],
185            buf_idx: 0,
186        };
187        Ok(reader)
188    }
189
190    /// Creates a `Reader` given a reader `Schema` and something implementing the `tokio::io::AsyncRead` trait
191    /// to read from.
192    ///
193    /// **NOTE** The avro header is going to be read automatically upon creation of the `Reader`.
194    pub fn with_schema(reader_schema: &Schema, mut inner: R) -> Result<Reader<R>, AvroError> {
195        let header = Header::from_reader(&mut inner)?;
196
197        let writer_schema = &header.writer_schema;
198        let resolved_schema = if reader_schema.fingerprint(&digest::SHA256).bytes
199            != writer_schema.fingerprint(&digest::SHA256).bytes
200        {
201            Some(resolve_schemas(writer_schema, reader_schema)?)
202        } else {
203            None
204        };
205
206        Ok(Reader {
207            header,
208            errored: false,
209            resolved_schema,
210            inner,
211            messages_remaining: 0,
212            buf: vec![],
213            buf_idx: 0,
214        })
215    }
216
217    /// Get a reference to the writer `Schema`.
218    pub fn writer_schema(&self) -> &Schema {
219        &self.header.writer_schema
220    }
221
222    /// Get a reference to the resolved schema
223    /// (or just the writer schema, if no reader schema was provided
224    ///  or the two schemas are identical)
225    pub fn schema(&self) -> &Schema {
226        match &self.resolved_schema {
227            Some(schema) => schema,
228            None => self.writer_schema(),
229        }
230    }
231
232    #[inline]
233    /// Read the next Avro value from the file, if one exists.
234    pub fn read_next(&mut self) -> Result<Option<Value>, AvroError> {
235        if self.is_empty() {
236            self.read_block_next()?;
237            if self.is_empty() {
238                return Ok(None);
239            }
240        }
241
242        let mut block_bytes = &self.buf[self.buf_idx..];
243        let b_original = block_bytes.len();
244        let schema = self.schema();
245        let item = from_avro_datum(schema, &mut block_bytes)?;
246        self.buf_idx += b_original - block_bytes.len();
247        self.messages_remaining -= 1;
248        Ok(Some(item))
249    }
250
251    fn is_empty(&self) -> bool {
252        self.messages_remaining == 0
253    }
254
255    fn fill_buf(&mut self, n: usize) -> Result<(), AvroError> {
256        // We don't have enough space in the buffer, need to grow it.
257        if n >= self.buf.len() {
258            self.buf.resize(n, 0);
259        }
260
261        self.inner.read_exact(&mut self.buf[..n])?;
262        // Cut the buffer down to exactly this block's payload. The resize above
263        // only ever grows, so a block shorter than a previous one would otherwise
264        // leave that block's tail visible past its own payload, and everything
265        // downstream reads the buffer by its length: `read_next` slices
266        // `self.buf[self.buf_idx..]`, and `Codec::decompress` hands the whole
267        // buffer to the decompressor. A block whose declared object count outruns
268        // its own bytes would then decode stale bytes as values instead of hitting
269        // the end of the block. `truncate` keeps the allocation, so the buffer is
270        // still reused across blocks.
271        self.buf.truncate(n);
272        self.buf_idx = 0;
273        Ok(())
274    }
275
276    fn read_block_next(&mut self) -> Result<(), AvroError> {
277        assert!(self.is_empty(), "Expected self to be empty!");
278        match util::read_long(&mut self.inner) {
279            Ok(block_len) => {
280                // The object count is read straight from the wire; cap it like
281                // every other wire-read length (and reject negatives, which wrap
282                // to a huge `usize`). Otherwise a crafted block with a huge count
283                // and a zero-byte schema (e.g. `null`) makes the reader spin
284                // decoding billions of empty values. Found by the reader_decode
285                // cargo-fuzz target.
286                self.messages_remaining = util::safe_len(block_len as usize)?;
287                let block_bytes = util::safe_len(util::read_long(&mut self.inner)? as usize)?;
288                self.fill_buf(block_bytes)?;
289                let mut marker = [0u8; 16];
290                self.inner.read_exact(&mut marker)?;
291
292                if marker != self.header.marker {
293                    return Err(DecodeError::MismatchedBlockHeader {
294                        expected: self.header.marker,
295                        actual: marker,
296                    }
297                    .into());
298                }
299
300                // NOTE (JAB): This doesn't fit this Reader pattern very well.
301                // `self.buf` is a growable buffer that is reused as the reader is iterated.
302                // For non `Codec::Null` variants, `decompress` will allocate a new `Vec`
303                // and replace `buf` with the new one, instead of reusing the same buffer.
304                // We can address this by using some "limited read" type to decode directly
305                // into the buffer. But this is fine, for now.
306                self.header.codec.decompress(&mut self.buf)?;
307
308                // `safe_len` above bounds the object count only by
309                // `MAX_ALLOCATION_BYTES`, which says nothing about the block that
310                // is supposed to contain those objects: a zero-width schema
311                // encodes every object to no bytes, so a handful of wire bytes can
312                // claim hundreds of millions of them and the reader will decode
313                // every one. Bound it against the payload now that the payload is
314                // known, which is only here, since the declared byte size is the
315                // *compressed* size.
316                let count = self.messages_remaining;
317                let payload_len = self.buf.len();
318                bound_block_object_count(self.schema().top_node(), count, payload_len)?;
319
320                Ok(())
321            }
322            Err(e) => {
323                if let AvroError::IO(std::io::ErrorKind::UnexpectedEof) = e {
324                    // to not return any error in case we only finished to read cleanly from the stream
325                    Ok(())
326                } else {
327                    Err(e)
328                }
329            }
330        }
331    }
332}
333
334impl<R: AvroRead> Iterator for Reader<R> {
335    type Item = Result<Value, AvroError>;
336
337    fn next(&mut self) -> Option<Self::Item> {
338        // to prevent continuing to read after the first error occurs
339        if self.errored {
340            return None;
341        };
342        match self.read_next() {
343            Ok(opt) => opt.map(Ok),
344            Err(e) => {
345                self.errored = true;
346                Some(Err(e))
347            }
348        }
349    }
350}
351
352pub struct SchemaResolver<'a> {
353    pub named: Vec<Option<NamedSchemaPiece>>,
354    pub indices: BTreeMap<FullName, usize>,
355    pub human_readable_field_path: Vec<String>,
356    pub current_human_readable_path_start: usize,
357    pub writer_to_reader_names: BTreeMap<usize, usize>,
358    pub reader_to_writer_names: BTreeMap<usize, usize>,
359    pub reader_to_resolved_names: BTreeMap<usize, usize>,
360    #[allow(dead_code)]
361    pub reader_fullnames: BTreeMap<usize, &'a FullName>,
362    pub reader_schema: &'a Schema,
363}
364
365impl<'a> SchemaResolver<'a> {
366    fn resolve_named(
367        &mut self,
368        writer: &Schema,
369        reader: &Schema,
370        writer_index: usize,
371        reader_index: usize,
372    ) -> Result<SchemaPiece, AvroError> {
373        let ws = writer.lookup(writer_index);
374        let rs = reader.lookup(reader_index);
375        let typ = match (&ws.piece, &rs.piece) {
376            (
377                SchemaPiece::Record {
378                    fields: w_fields,
379                    lookup: w_lookup,
380                    ..
381                },
382                SchemaPiece::Record {
383                    fields: r_fields,
384                    lookup: _r_lookup,
385                    ..
386                },
387            ) => {
388                let mut defaults = Vec::new();
389                let mut fields: Vec<Option<RecordField>> = Vec::new();
390                for (r_index, rf) in r_fields.iter().enumerate() {
391                    match w_lookup.get(&rf.name) {
392                        None => {
393                            let default_field = match &rf.default {
394                                Some(v) => ResolvedDefaultValueField {
395                                    name: rf.name.clone(),
396                                    doc: rf.doc.clone(),
397                                    default: reader
398                                        .top_node_or_named()
399                                        .step(&rf.schema)
400                                        .lookup()
401                                        .json_to_value(v)?,
402                                    order: rf.order.clone(),
403                                    position: r_index,
404                                },
405                                None => return Err(SchemaResolutionError::new(format!(
406                                    "Reader field `{}.{}` not found in writer, and has no default",
407                                    self.get_current_human_readable_path(),
408                                    rf.name
409                                ))
410                                .into()),
411                            };
412                            defaults.push(default_field);
413                        }
414                        Some(w_index) => {
415                            if fields.len() > *w_index && fields[*w_index].is_some() {
416                                return Err(SchemaResolutionError::new(format!(
417                                    "Duplicate field `{}.{}` in schema",
418                                    self.get_current_human_readable_path(),
419                                    rf.name
420                                ))
421                                .into());
422                            }
423                            let wf = &w_fields[*w_index];
424                            let w_node = SchemaNodeOrNamed {
425                                root: writer,
426                                inner: wf.schema.as_ref(),
427                            };
428                            let r_node = SchemaNodeOrNamed {
429                                root: reader,
430                                inner: rf.schema.as_ref(),
431                            };
432
433                            self.human_readable_field_path.push(rf.name.clone());
434                            let new_inner = self.resolve(w_node, r_node)?;
435                            self.human_readable_field_path.pop();
436
437                            let field = RecordField {
438                                name: rf.name.clone(),
439                                doc: rf.doc.clone(),
440                                default: rf.default.clone(),
441                                schema: new_inner,
442                                order: rf.order.clone(),
443                                position: r_index,
444                            };
445                            while fields.len() <= *w_index {
446                                fields.push(None);
447                            }
448                            fields[*w_index] = Some(field)
449                        }
450                    }
451                }
452                while fields.len() < w_fields.len() {
453                    fields.push(None);
454                }
455                let mut n_present = 0;
456                let fields = fields
457                    .into_iter()
458                    .enumerate()
459                    .map(|(i, rf)| match rf {
460                        Some(rf) => {
461                            n_present += 1;
462                            ResolvedRecordField::Present(rf)
463                        }
464                        None => {
465                            // Clone the chunk of the writer schema appearing here.
466                            // We could probably be clever and avoid some cloning,
467                            // but absolute highest performance probably isn't important for schema resolution.
468                            //
469                            // The cloned writer schema piece is needed to guide decoding of the value,
470                            // since even though it doesn't appear in the reader schema it needs
471                            // to be decoded to know where it ends.
472                            //
473                            // TODO -- We could try to come up with a "Dummy" schema variant
474                            // that does only enough decoding to find the end of a value,
475                            // and maybe save some time.
476                            let writer_schema_piece = SchemaNodeOrNamed {
477                                root: writer,
478                                inner: w_fields[i].schema.as_ref(),
479                            }
480                            .to_schema();
481                            ResolvedRecordField::Absent(writer_schema_piece)
482                        }
483                    })
484                    .collect();
485                let n_reader_fields = defaults.len() + n_present;
486                SchemaPiece::ResolveRecord {
487                    defaults,
488                    fields,
489                    n_reader_fields,
490                }
491            }
492            (
493                SchemaPiece::Enum {
494                    symbols: w_symbols, ..
495                },
496                SchemaPiece::Enum {
497                    symbols: r_symbols,
498                    doc,
499                    default_idx,
500                },
501            ) => {
502                let r_map = r_symbols
503                    .iter()
504                    .enumerate()
505                    .map(|(i, s)| (s, i))
506                    .collect::<BTreeMap<_, _>>();
507                let symbols = w_symbols
508                    .iter()
509                    .map(|s| {
510                        r_map
511                            .get(s)
512                            .map(|i| (*i, s.clone()))
513                            .ok_or_else(|| s.clone())
514                    })
515                    .collect();
516                SchemaPiece::ResolveEnum {
517                    doc: doc.clone(),
518                    symbols,
519                    default: default_idx.map(|i| (i, r_symbols[i].clone())),
520                }
521            }
522            (SchemaPiece::Fixed { size: wsz }, SchemaPiece::Fixed { size: rsz }) => {
523                if *wsz == *rsz {
524                    SchemaPiece::Fixed { size: *wsz }
525                } else {
526                    return Err(SchemaResolutionError::new(format!(
527                        "Fixed schema {:?}: sizes don't match ({}, {}) for field `{}`",
528                        &rs.name,
529                        wsz,
530                        rsz,
531                        self.get_current_human_readable_path(),
532                    ))
533                    .into());
534                }
535            }
536            (
537                SchemaPiece::Decimal {
538                    precision: wp,
539                    scale: wscale,
540                    fixed_size: wsz,
541                },
542                SchemaPiece::Decimal {
543                    precision: rp,
544                    scale: rscale,
545                    fixed_size: rsz,
546                },
547            ) => {
548                if wp != rp {
549                    return Err(SchemaResolutionError::new(format!(
550                        "Decimal schema {:?}: precisions don't match: {}, {} for field `{}`",
551                        &rs.name,
552                        wp,
553                        rp,
554                        self.get_current_human_readable_path(),
555                    ))
556                    .into());
557                }
558                if wscale != rscale {
559                    return Err(SchemaResolutionError::new(format!(
560                        "Decimal schema {:?}: sizes don't match: {}, {} for field `{}`",
561                        &rs.name,
562                        wscale,
563                        rscale,
564                        self.get_current_human_readable_path(),
565                    ))
566                    .into());
567                }
568                if wsz != rsz {
569                    return Err(SchemaResolutionError::new(format!(
570                        "Decimal schema {:?}: sizes don't match: {:?}, {:?} for field `{}`",
571                        &rs.name,
572                        wsz,
573                        rsz,
574                        self.get_current_human_readable_path(),
575                    ))
576                    .into());
577                }
578                SchemaPiece::Decimal {
579                    precision: *wp,
580                    scale: *wscale,
581                    fixed_size: *wsz,
582                }
583            }
584            (SchemaPiece::Decimal { fixed_size, .. }, SchemaPiece::Fixed { size })
585                if *fixed_size == Some(*size) =>
586            {
587                SchemaPiece::Fixed { size: *size }
588            }
589            (
590                SchemaPiece::Fixed { size },
591                SchemaPiece::Decimal {
592                    precision,
593                    scale,
594                    fixed_size,
595                },
596            ) if *fixed_size == Some(*size) => SchemaPiece::Decimal {
597                precision: *precision,
598                scale: *scale,
599                fixed_size: *fixed_size,
600            },
601
602            (_, SchemaPiece::ResolveRecord { .. })
603            | (_, SchemaPiece::ResolveEnum { .. })
604            | (SchemaPiece::ResolveRecord { .. }, _)
605            | (SchemaPiece::ResolveEnum { .. }, _) => {
606                return Err(SchemaResolutionError::new(
607                    "Attempted to resolve an already resolved schema".to_string(),
608                )
609                .into());
610            }
611
612            (_wt, _rt) => {
613                return Err(SchemaResolutionError::new(format!(
614                    "Non-matching schemas: writer: {:?}, reader: {:?}",
615                    ws.name, rs.name
616                ))
617                .into());
618            }
619        };
620        Ok(typ)
621    }
622
623    pub fn resolve(
624        &mut self,
625        writer: SchemaNodeOrNamed,
626        reader: SchemaNodeOrNamed,
627    ) -> Result<SchemaPieceOrNamed, AvroError> {
628        let previous_human_readable_path_start = self.current_human_readable_path_start;
629        let (_, named_node) = reader.inner.get_piece_and_name(reader.root);
630        if let Some(full_name) = named_node {
631            self.current_human_readable_path_start = self.human_readable_field_path.len();
632            self.human_readable_field_path.push(full_name.human_name());
633        }
634
635        let inner = match (writer.inner, reader.inner) {
636            // Both schemas are unions - the most complicated case, but simpler than it looks.
637            // For each variant in the writer, we attempt to find a matching variant in the reader,
638            // either by type (for anonymous nodes) or by name (for named nodes).
639            //
640            // Having found a match, we resolve the writer variant against the reader variant,
641            // and record it in the resolved node.
642            //
643            // If either no match is found, or resolution on the matches fails, it is not an error
644            // -- it simply means that the corresponding entry in `permutation` will be `None`,
645            // and reading will fail if that variant is expressed. But
646            // reading variants that *do* match and resolve will still be possible.
647            //
648            // See the doc comment on `SchemaPiece::ResolveUnionUnion` for an explanation of the format of `permutation`.
649            (
650                SchemaPieceRefOrNamed::Piece(SchemaPiece::Union(w_inner)),
651                SchemaPieceRefOrNamed::Piece(SchemaPiece::Union(r_inner)),
652            ) => {
653                let w2r = self.writer_to_reader_names.clone();
654                // permutation[1] is Some((j, val)) iff the i'th writer variant
655                // _matches_ the j'th reader variant
656                // (i.e., it is the same primitive type, or the same kind of named type and has the same name, or a decimal with the same parameters)
657                // and successfully _resolves_ against it,
658                // and None otherwise.
659                //
660                // An example of types that match but don't resolve would be two records with the same name but incompatible fields.
661                let permutation = w_inner
662                    .variants()
663                    .iter()
664                    .map(|w_variant| {
665                        let (r_idx, r_variant) =
666                            r_inner.match_promote_writer(w_variant, &w2r).ok_or_else(|| {
667                                SchemaResolutionError::new(format!(
668                                    "Failed to match writer union variant `{}` against any variant in the reader for field `{}`",
669                                    w_variant.get_human_name(writer.root),
670                                    self.get_current_human_readable_path()
671                                ))
672                            })?;
673                        let resolved =
674                            self.resolve(writer.step(w_variant), reader.step(r_variant))?;
675                        Ok((r_idx, resolved))
676                    })
677                    .collect();
678                let n_reader_variants = r_inner.variants().len();
679                let reader_null_variant = r_inner
680                    .variants()
681                    .iter()
682                    .position(|v| v == &SchemaPieceOrNamed::Piece(SchemaPiece::Null));
683                SchemaPieceOrNamed::Piece(SchemaPiece::ResolveUnionUnion {
684                    permutation,
685                    n_reader_variants,
686                    reader_null_variant,
687                })
688            }
689            // Writer is concrete; reader is union
690            (other, SchemaPieceRefOrNamed::Piece(SchemaPiece::Union(r_inner))) => {
691                let n_reader_variants = r_inner.variants().len();
692                let reader_null_variant = r_inner
693                    .variants()
694                    .iter()
695                    .position(|v| v == &SchemaPieceOrNamed::Piece(SchemaPiece::Null));
696                let (index, r_inner) = r_inner
697                    .match_ref_promote_writer(other, &self.writer_to_reader_names)
698                    .ok_or_else(|| {
699                        SchemaResolutionError::new(
700                            format!("No matching schema in reader union for writer type `{}` for field `{}`",
701                                    other.get_human_name(writer.root),
702                                    self.get_current_human_readable_path()))
703                    })?;
704                let inner = Box::new(self.resolve(writer.step_ref(other), reader.step(r_inner))?);
705                SchemaPieceOrNamed::Piece(SchemaPiece::ResolveConcreteUnion {
706                    index,
707                    inner,
708                    n_reader_variants,
709                    reader_null_variant,
710                })
711            }
712            // Writer is union; reader is concrete
713            (SchemaPieceRefOrNamed::Piece(SchemaPiece::Union(w_inner)), other) => {
714                let (index, w_inner) = w_inner
715                    .match_ref_promote_reader(other, &self.reader_to_writer_names)
716                    .ok_or_else(|| {
717                        // `other` is the reader's concrete node (the second element
718                        // of the match), so its name must be looked up in the
719                        // reader's schema. Using `writer.root` here indexed the
720                        // writer's (possibly empty) `named` table out of bounds.
721                        SchemaResolutionError::new(
722                            format!("No matching schema in writer union for reader type `{}` for field `{}`",
723                                    other.get_human_name(reader.root),
724                                    self.get_current_human_readable_path()))
725                    })?;
726                let inner = Box::new(self.resolve(writer.step(w_inner), reader.step_ref(other))?);
727                SchemaPieceOrNamed::Piece(SchemaPiece::ResolveUnionConcrete { index, inner })
728            }
729            // Any other anonymous type.
730            (SchemaPieceRefOrNamed::Piece(wp), SchemaPieceRefOrNamed::Piece(rp)) => {
731                match (wp, rp) {
732                    // Normally for types that are underlyingly "long", we just interpret them according to the reader schema.
733                    // In this special case, it is better to interpret them according to the _writer_ schema:
734                    // By treating the written value as millis, we will decode the same DateTime values as were written.
735                    //
736                    // For example: if a writer wrote milliseconds and a reader tries to read it as microseconds,
737                    // it will be off by a factor of 1000 from the timestamp that the writer was intending to write
738                    (SchemaPiece::TimestampMilli, SchemaPiece::TimestampMicro) => {
739                        SchemaPieceOrNamed::Piece(SchemaPiece::TimestampMilli)
740                    }
741                    // See above
742                    (SchemaPiece::TimestampMicro, SchemaPiece::TimestampMilli) => {
743                        SchemaPieceOrNamed::Piece(SchemaPiece::TimestampMicro)
744                    }
745                    (SchemaPiece::Date, SchemaPiece::TimestampMilli)
746                    | (SchemaPiece::Date, SchemaPiece::TimestampMicro) => {
747                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveDateTimestamp)
748                    }
749                    (wp, rp) if wp.is_underlying_int() && rp.is_underlying_int() => {
750                        SchemaPieceOrNamed::Piece(rp.clone()) // This clone is just a copy - none of the underlying int/long types own heap memory.
751                    }
752                    (wp, rp) if wp.is_underlying_long() && rp.is_underlying_long() => {
753                        SchemaPieceOrNamed::Piece(rp.clone()) // see above comment
754                    }
755                    (wp, SchemaPiece::TimestampMilli) if wp.is_underlying_int() => {
756                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveIntTsMilli)
757                    }
758                    (wp, SchemaPiece::TimestampMicro) if wp.is_underlying_int() => {
759                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveIntTsMicro)
760                    }
761                    (SchemaPiece::Null, SchemaPiece::Null) => {
762                        SchemaPieceOrNamed::Piece(SchemaPiece::Null)
763                    }
764                    (SchemaPiece::Boolean, SchemaPiece::Boolean) => {
765                        SchemaPieceOrNamed::Piece(SchemaPiece::Boolean)
766                    }
767                    (SchemaPiece::Int, SchemaPiece::Long) => {
768                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveIntLong)
769                    }
770                    (SchemaPiece::Int, SchemaPiece::Float) => {
771                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveIntFloat)
772                    }
773                    (SchemaPiece::Int, SchemaPiece::Double) => {
774                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveIntDouble)
775                    }
776                    (SchemaPiece::Long, SchemaPiece::Float) => {
777                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveLongFloat)
778                    }
779                    (SchemaPiece::Long, SchemaPiece::Double) => {
780                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveLongDouble)
781                    }
782                    (SchemaPiece::Float, SchemaPiece::Float) => {
783                        SchemaPieceOrNamed::Piece(SchemaPiece::Float)
784                    }
785                    (SchemaPiece::Float, SchemaPiece::Double) => {
786                        SchemaPieceOrNamed::Piece(SchemaPiece::ResolveFloatDouble)
787                    }
788                    (SchemaPiece::Double, SchemaPiece::Double) => {
789                        SchemaPieceOrNamed::Piece(SchemaPiece::Double)
790                    }
791                    (b, SchemaPiece::Bytes)
792                        if b == &SchemaPiece::Bytes || b == &SchemaPiece::String =>
793                    {
794                        SchemaPieceOrNamed::Piece(SchemaPiece::Bytes)
795                    }
796                    (s, SchemaPiece::String)
797                        if s == &SchemaPiece::String || s == &SchemaPiece::Bytes =>
798                    {
799                        SchemaPieceOrNamed::Piece(SchemaPiece::String)
800                    }
801                    (SchemaPiece::Array(w_inner), SchemaPiece::Array(r_inner)) => {
802                        let inner =
803                            self.resolve(writer.step(&**w_inner), reader.step(&**r_inner))?;
804                        SchemaPieceOrNamed::Piece(SchemaPiece::Array(Box::new(inner)))
805                    }
806                    (SchemaPiece::Map(w_inner), SchemaPiece::Map(r_inner)) => {
807                        let inner =
808                            self.resolve(writer.step(&**w_inner), reader.step(&**r_inner))?;
809                        SchemaPieceOrNamed::Piece(SchemaPiece::Map(Box::new(inner)))
810                    }
811                    (
812                        SchemaPiece::Decimal {
813                            precision: wp,
814                            scale: ws,
815                            fixed_size: wf,
816                        },
817                        SchemaPiece::Decimal {
818                            precision: rp,
819                            scale: rs,
820                            fixed_size: rf,
821                        },
822                    ) => {
823                        if wp == rp && ws == rs && wf == rf {
824                            SchemaPieceOrNamed::Piece(SchemaPiece::Decimal {
825                                precision: *wp,
826                                scale: *ws,
827                                fixed_size: *wf,
828                            })
829                        } else {
830                            return Err(SchemaResolutionError::new(format!(
831                                "Decimal types must match in precision, scale, and fixed size. \
832                                Got ({:?}, {:?}, {:?}); ({:?}, {:?}. {:?}) for field `{}`",
833                                wp,
834                                ws,
835                                wf,
836                                rp,
837                                rs,
838                                rf,
839                                self.get_current_human_readable_path(),
840                            ))
841                            .into());
842                        }
843                    }
844                    (SchemaPiece::Decimal { fixed_size, .. }, SchemaPiece::Bytes)
845                        if *fixed_size == None =>
846                    {
847                        SchemaPieceOrNamed::Piece(SchemaPiece::Bytes)
848                    }
849                    // TODO [btv] We probably want to rethink what we're doing here, rather than just add
850                    // a new branch for every possible "logical" type. Perhaps logical types with the
851                    // same underlying type should always be resolvable to the reader schema's type?
852                    (SchemaPiece::Json, SchemaPiece::Json) => {
853                        SchemaPieceOrNamed::Piece(SchemaPiece::Json)
854                    }
855                    (SchemaPiece::Uuid, SchemaPiece::Uuid) => {
856                        SchemaPieceOrNamed::Piece(SchemaPiece::Uuid)
857                    }
858                    (
859                        SchemaPiece::Bytes,
860                        SchemaPiece::Decimal {
861                            precision,
862                            scale,
863                            fixed_size,
864                        },
865                    ) if *fixed_size == None => SchemaPieceOrNamed::Piece(SchemaPiece::Decimal {
866                        precision: *precision,
867                        scale: *scale,
868                        fixed_size: *fixed_size,
869                    }),
870                    (ws, rs) => {
871                        return Err(SchemaResolutionError::new(format!(
872                            "Writer schema has type `{:?}`, but reader schema has type `{:?}` for field `{}`",
873                            ws,
874                            rs,
875                            self.get_current_human_readable_path(),
876                        ))
877                        .into());
878                    }
879                }
880            }
881            // Named types
882            (SchemaPieceRefOrNamed::Named(w_index), SchemaPieceRefOrNamed::Named(r_index)) => {
883                if self.writer_to_reader_names.get(&w_index) != Some(&r_index) {
884                    // The nodes in the two schemas have different names. Resolution fails.
885                    let (w_name, r_name) = (
886                        &writer.root.lookup(w_index).name,
887                        &reader.root.lookup(r_index).name,
888                    );
889                    return Err(SchemaResolutionError::new(format!("Attempted to resolve writer schema node named {:?} against reader schema node named {:?}", w_name, r_name)).into());
890                }
891                // Check if we have already resolved the name previously, and if so, return a reference to
892                // it (in the new schema's namespace).
893                let idx = match self.reader_to_resolved_names.get(&r_index) {
894                    Some(resolved) => *resolved,
895                    None => {
896                        // We have not resolved this name yet; do so, and record it in the set of named schemas.
897                        // We need to push a placeholder beforehand, because schemas can be recursive;
898                        // a schema nested under this one may reference it.
899                        // A plausible example: {"type": "record", "name": "linked_list", "fields": [{"name": "next", "type": ["null", "linked_list"]}]}
900                        // Thus, `self.reader_to_resolved_names` needs to be correct for this node's index *before* we traverse the nodes under it.
901                        let resolved_idx = self.named.len();
902                        self.reader_to_resolved_names.insert(r_index, resolved_idx);
903                        self.named.push(None);
904                        let piece =
905                            match self.resolve_named(writer.root, reader.root, w_index, r_index) {
906                                Ok(piece) => piece,
907                                Err(e) => {
908                                    // Roll back to the state before this node. We must remove not
909                                    // only this node's placeholder but also any nested named nodes
910                                    // resolved while resolving it: they live at indices
911                                    // `>= resolved_idx` and are unreachable now that this resolution
912                                    // failed. A plain `pop()` removed only the last one, orphaning a
913                                    // `None` placeholder (which a later `Option::unwrap` panics on)
914                                    // whenever a nested node had been pushed. Union resolution stores
915                                    // rather than propagates this error, so the orphan would survive.
916                                    self.named.truncate(resolved_idx);
917                                    self.reader_to_resolved_names
918                                        .retain(|_, v| *v < resolved_idx);
919                                    self.indices.retain(|_, v| *v < resolved_idx);
920                                    return Err(e);
921                                }
922                            };
923                        let name = &self.reader_schema.named[r_index].name;
924                        let ns = NamedSchemaPiece {
925                            name: name.clone(),
926                            piece,
927                        };
928                        self.named[resolved_idx] = Some(ns);
929                        self.indices.insert(name.clone(), resolved_idx);
930
931                        resolved_idx
932                    }
933                };
934                SchemaPieceOrNamed::Named(idx)
935            }
936            (ws, rs) => {
937                return Err(SchemaResolutionError::new(format!(
938                    "Schemas don't match: {:?}, {:?} for field `{}`",
939                    ws.get_piece_and_name(writer.root).0,
940                    rs.get_piece_and_name(reader.root).0,
941                    self.get_current_human_readable_path(),
942                ))
943                .into());
944            }
945        };
946        if named_node.is_some() {
947            self.human_readable_field_path.pop();
948            self.current_human_readable_path_start = previous_human_readable_path_start;
949        }
950        Ok(inner)
951    }
952
953    fn get_current_human_readable_path(&self) -> String {
954        self.human_readable_field_path[self.current_human_readable_path_start..].join(".")
955    }
956}
957
958/// Decode a `Value` encoded in Avro format given its `Schema` and anything implementing `io::Read`
959/// to read from.
960///
961/// In case a reader `Schema` is provided, schema resolution will also be performed.
962///
963/// **NOTE** This function has a quite small niche of usage and does NOT take care of reading the
964/// header and consecutive data blocks; use [`Reader`](struct.Reader.html) if you don't know what
965/// you are doing, instead.
966pub fn from_avro_datum<R: AvroRead>(schema: &Schema, reader: &mut R) -> Result<Value, AvroError> {
967    let value = decode(schema.top_node(), reader)?;
968    Ok(value)
969}
970
971#[cfg(test)]
972mod tests {
973    use std::io::Cursor;
974
975    use mz_ore::assert_err;
976
977    use crate::Reader;
978    use crate::types::{Record, ToAvro};
979
980    use super::*;
981
982    /// Assemble an object-container file: header (writer schema, `null` codec,
983    /// sync marker) followed by `blocks`. Each block is given as `(declared
984    /// count, declared byte size, payload)` so a test can lie about the framing
985    /// the way a corrupt or hostile file does.
986    fn ocf(schema_json: &str, blocks: &[(i64, i64, &[u8])]) -> Vec<u8> {
987        fn blob(bytes: &[u8], out: &mut Vec<u8>) {
988            util::zig_i64(bytes.len() as i64, out);
989            out.extend_from_slice(bytes);
990        }
991
992        let marker = [7u8; 16];
993        let mut out = b"Obj\x01".to_vec();
994        util::zig_i64(2, &mut out); // metadata map: one block of two entries
995        blob(b"avro.schema", &mut out);
996        blob(schema_json.as_bytes(), &mut out);
997        blob(b"avro.codec", &mut out);
998        blob(b"null", &mut out);
999        util::zig_i64(0, &mut out); // end of metadata map
1000        out.extend_from_slice(&marker);
1001        for (count, size, payload) in blocks {
1002            util::zig_i64(*count, &mut out);
1003            util::zig_i64(*size, &mut out);
1004            out.extend_from_slice(payload);
1005            out.extend_from_slice(&marker);
1006        }
1007        out
1008    }
1009
1010    /// A record of only `null` fields: valid, and encodes to zero bytes.
1011    const ZERO_WIDTH_SCHEMA: &str =
1012        r#"{"type":"record","name":"R","fields":[{"name":"g","type":"null"}]}"#;
1013
1014    #[mz_ore::test]
1015    fn reader_does_not_decode_a_previous_blocks_bytes() {
1016        // One buffer is reused across blocks, so a block shorter than its
1017        // predecessor must not leave that predecessor's tail readable. Block 2
1018        // here declares two objects but carries only one, and its payload is
1019        // shorter than block 1's, whose bytes are arranged so that what lands
1020        // past block 2's payload would decode as a perfectly good `"x"`. The
1021        // second decode has to run out of input instead.
1022        //
1023        // The block bound does not catch this: `string` has a one-byte floor and
1024        // the block claims 2 objects in 3 bytes.
1025        let long = &[0x0a, b'A', b'B', 0x02, b'x', b'C'][..]; // one 5-char string
1026        let short = &[0x04, b'a', b'b'][..]; // one 2-char string
1027        let file = ocf(
1028            r#""string""#,
1029            &[(1, long.len() as i64, long), (2, short.len() as i64, short)],
1030        );
1031
1032        let items: Vec<_> = Reader::new(&file[..]).expect("OCF header parses").collect();
1033        assert_eq!(items.len(), 3, "expected two values then an error");
1034        assert_eq!(
1035            items[1].as_ref().expect("block 2's real value decodes"),
1036            &Value::String("ab".into())
1037        );
1038        assert_err!(&items[2]);
1039    }
1040
1041    #[mz_ore::test]
1042    fn reader_rejects_zero_width_block_count_within_allocation_budget() {
1043        // `safe_len` accepts any count up to `MAX_ALLOCATION_BYTES`, and a
1044        // zero-width object consumes no input, so nothing but the block bound
1045        // stops a 0-byte payload from claiming 100M objects and being believed.
1046        // Regression for a reader_decode cargo-fuzz timeout.
1047        let file = ocf(ZERO_WIDTH_SCHEMA, &[(100_000_000, 0, &[])]);
1048        let err = Reader::new(&file[..])
1049            .expect("OCF header parses")
1050            .find_map(|item| item.err())
1051            .expect("the oversized count must be rejected");
1052        assert!(
1053            err.to_string().contains("exceeds limit"),
1054            "unexpected error: {err}"
1055        );
1056    }
1057
1058    #[mz_ore::test]
1059    fn reader_accepts_honest_zero_width_block() {
1060        // The bound above must not reject a real zero-width block: the whole
1061        // point of the node weighting is that a byte count cannot judge one.
1062        let file = ocf(ZERO_WIDTH_SCHEMA, &[(3, 0, &[])]);
1063        let values: Vec<Value> = Reader::new(&file[..])
1064            .expect("OCF header parses")
1065            .collect::<Result<_, _>>()
1066            .expect("an honest zero-width block must decode");
1067        assert_eq!(values.len(), 3);
1068    }
1069
1070    #[mz_ore::test]
1071    fn reader_rejects_block_count_exceeding_payload() {
1072        // A `string` occupies at least its one-byte length varint, so 100 of them
1073        // cannot fit in three bytes however the payload is arranged.
1074        let payload = &[0x04, b'a', b'b'][..];
1075        let file = ocf(r#""string""#, &[(100, payload.len() as i64, payload)]);
1076        let err = Reader::new(&file[..])
1077            .expect("OCF header parses")
1078            .find_map(|item| item.err())
1079            .expect("a count larger than the payload must be rejected");
1080        assert!(
1081            err.to_string().contains("exceeds block payload"),
1082            "unexpected error: {err}"
1083        );
1084    }
1085
1086    #[mz_ore::test]
1087    fn reader_rejects_huge_block_object_count() {
1088        // A crafted object-container block with a huge object count and a
1089        // zero-byte (`null`) schema must be rejected, not spin decoding billions
1090        // of empty values. Regression for the reader_decode cargo-fuzz timeout.
1091        let bytes: &[u8] = &[
1092            0x4f, 0x62, 0x6a, 0x01, 0x04, 0x16, 0x61, 0x76, 0x72, 0x6f, 0x2e, 0x73, 0x63, 0x68,
1093            0x65, 0x6d, 0x61, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x22,
1094            0x6e, 0x75, 0x6c, 0x6c, 0x22, 0x20, 0x00, 0x00, 0x00, 0x64, 0x64, 0x64, 0x64, 0x64,
1095            0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0xbf, 0x00, 0x26,
1096            0x35, 0x33, 0x39, 0x33, 0x34, 0x38, 0x33, 0xcd, 0x45, 0x38, 0x56, 0xb1, 0x00, 0x00,
1097            0x64, 0x64, 0x7a, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64,
1098            0x64, 0x64, 0x64, 0x64, 0x64, 0x64, 0x64,
1099        ];
1100        let reader = Reader::new(bytes).expect("OCF header parses");
1101        // Iteration must terminate with an error (the oversized count is
1102        // rejected), not hang.
1103        assert!(
1104            reader.into_iter().any(|item| item.is_err()),
1105            "expected a decode error from the oversized block count"
1106        );
1107    }
1108
1109    static SCHEMA: &str = r#"
1110            {
1111                "type": "record",
1112                "name": "test",
1113                "fields": [
1114                    {"name": "a", "type": "long", "default": 42},
1115                    {"name": "b", "type": "string"}
1116                ]
1117            }
1118        "#;
1119    static UNION_SCHEMA: &str = r#"
1120            ["null", "long"]
1121        "#;
1122    static ENCODED: &[u8] = &[
1123        79u8, 98u8, 106u8, 1u8, 4u8, 22u8, 97u8, 118u8, 114u8, 111u8, 46u8, 115u8, 99u8, 104u8,
1124        101u8, 109u8, 97u8, 222u8, 1u8, 123u8, 34u8, 116u8, 121u8, 112u8, 101u8, 34u8, 58u8, 34u8,
1125        114u8, 101u8, 99u8, 111u8, 114u8, 100u8, 34u8, 44u8, 34u8, 110u8, 97u8, 109u8, 101u8, 34u8,
1126        58u8, 34u8, 116u8, 101u8, 115u8, 116u8, 34u8, 44u8, 34u8, 102u8, 105u8, 101u8, 108u8,
1127        100u8, 115u8, 34u8, 58u8, 91u8, 123u8, 34u8, 110u8, 97u8, 109u8, 101u8, 34u8, 58u8, 34u8,
1128        97u8, 34u8, 44u8, 34u8, 116u8, 121u8, 112u8, 101u8, 34u8, 58u8, 34u8, 108u8, 111u8, 110u8,
1129        103u8, 34u8, 44u8, 34u8, 100u8, 101u8, 102u8, 97u8, 117u8, 108u8, 116u8, 34u8, 58u8, 52u8,
1130        50u8, 125u8, 44u8, 123u8, 34u8, 110u8, 97u8, 109u8, 101u8, 34u8, 58u8, 34u8, 98u8, 34u8,
1131        44u8, 34u8, 116u8, 121u8, 112u8, 101u8, 34u8, 58u8, 34u8, 115u8, 116u8, 114u8, 105u8,
1132        110u8, 103u8, 34u8, 125u8, 93u8, 125u8, 20u8, 97u8, 118u8, 114u8, 111u8, 46u8, 99u8, 111u8,
1133        100u8, 101u8, 99u8, 8u8, 110u8, 117u8, 108u8, 108u8, 0u8, 94u8, 61u8, 54u8, 221u8, 190u8,
1134        207u8, 108u8, 180u8, 158u8, 57u8, 114u8, 40u8, 173u8, 199u8, 228u8, 239u8, 4u8, 20u8, 54u8,
1135        6u8, 102u8, 111u8, 111u8, 84u8, 6u8, 98u8, 97u8, 114u8, 94u8, 61u8, 54u8, 221u8, 190u8,
1136        207u8, 108u8, 180u8, 158u8, 57u8, 114u8, 40u8, 173u8, 199u8, 228u8, 239u8,
1137    ];
1138
1139    #[mz_ore::test]
1140    fn test_from_avro_datum() {
1141        let schema: Schema = SCHEMA.parse().unwrap();
1142        let mut encoded: &'static [u8] = &[54, 6, 102, 111, 111];
1143
1144        let mut record = Record::new(schema.top_node()).unwrap();
1145        record.put("a", 27i64);
1146        record.put("b", "foo");
1147        let expected = record.avro();
1148
1149        assert_eq!(from_avro_datum(&schema, &mut encoded).unwrap(), expected);
1150    }
1151
1152    #[mz_ore::test]
1153    fn test_null_union() {
1154        let schema: Schema = UNION_SCHEMA.parse().unwrap();
1155        let mut encoded: &'static [u8] = &[2, 0];
1156
1157        assert_eq!(
1158            from_avro_datum(&schema, &mut encoded).unwrap(),
1159            Value::Union {
1160                index: 1,
1161                inner: Box::new(Value::Long(0)),
1162                n_variants: 2,
1163                null_variant: Some(0)
1164            }
1165        );
1166    }
1167
1168    #[mz_ore::test]
1169    #[cfg_attr(miri, ignore)] // unsupported operation: inline assembly is not supported
1170    fn test_reader_stream() {
1171        let schema: Schema = SCHEMA.parse().unwrap();
1172        let reader = Reader::with_schema(&schema, ENCODED).unwrap();
1173
1174        let mut record1 = Record::new(schema.top_node()).unwrap();
1175        record1.put("a", 27i64);
1176        record1.put("b", "foo");
1177
1178        let mut record2 = Record::new(schema.top_node()).unwrap();
1179        record2.put("a", 42i64);
1180        record2.put("b", "bar");
1181
1182        let expected = [record1.avro(), record2.avro()];
1183
1184        for (i, value) in reader.enumerate() {
1185            assert_eq!(value.unwrap(), expected[i]);
1186        }
1187    }
1188
1189    #[mz_ore::test]
1190    fn test_reader_invalid_header() {
1191        let schema: Schema = SCHEMA.parse().unwrap();
1192        let invalid = ENCODED.iter().skip(1).copied().collect::<Vec<u8>>();
1193        assert!(Reader::with_schema(&schema, &invalid[..]).is_err());
1194    }
1195
1196    #[mz_ore::test]
1197    #[cfg_attr(miri, ignore)] // unsupported operation: inline assembly is not supported
1198    fn test_reader_invalid_block() {
1199        let schema: Schema = SCHEMA.parse().unwrap();
1200        let invalid = ENCODED
1201            .iter()
1202            .rev()
1203            .skip(19)
1204            .copied()
1205            .collect::<Vec<u8>>()
1206            .into_iter()
1207            .rev()
1208            .collect::<Vec<u8>>();
1209        let reader = Reader::with_schema(&schema, &invalid[..]).unwrap();
1210        for value in reader {
1211            assert_err!(value);
1212        }
1213    }
1214
1215    #[mz_ore::test]
1216    fn test_reader_empty_buffer() {
1217        let empty = Cursor::new(Vec::new());
1218        assert!(Reader::new(empty).is_err());
1219    }
1220
1221    #[mz_ore::test]
1222    fn test_reader_only_header() {
1223        let invalid = ENCODED.iter().copied().take(165).collect::<Vec<u8>>();
1224        let reader = Reader::new(&invalid[..]).unwrap();
1225        for value in reader {
1226            assert_err!(value);
1227        }
1228    }
1229
1230    #[mz_ore::test]
1231    fn test_resolution_nested_types_error() {
1232        let r = r#"
1233{
1234    "type": "record",
1235    "name": "com.materialize.foo",
1236    "fields": [
1237        {"name": "f1", "type": {"type": "record", "name": "com.materialize.bar", "fields": [{"name": "f1_1", "type": "int"}]}}
1238    ]
1239}
1240"#;
1241        let w = r#"
1242{
1243    "type": "record",
1244    "name": "com.materialize.foo",
1245    "fields": [
1246        {"name": "f1", "type": {"type": "record", "name": "com.materialize.bar", "fields": [{"name": "f1_1", "type": "double"}]}}
1247    ]
1248}
1249"#;
1250        let r: Schema = r.parse().unwrap();
1251        let w: Schema = w.parse().unwrap();
1252        let err_str = if let Result::Err(AvroError::ResolveSchema(SchemaResolutionError(s))) =
1253            resolve_schemas(&w, &r)
1254        {
1255            s
1256        } else {
1257            panic!("Expected schema resolution failure");
1258        };
1259        // The field name here must NOT contain `com.materialize.foo`,
1260        // because explicitly named types are all relative to a global
1261        // namespace (i.e., they don't nest).
1262        assert_eq!(
1263            &err_str,
1264            "Writer schema has type `Double`, but reader schema has type `Int` for field `com.materialize.bar.f1_1`"
1265        );
1266    }
1267
1268    #[mz_ore::test]
1269    fn test_extra_fields_without_default_error() {
1270        let r = r#"
1271{
1272    "type": "record",
1273    "name": "com.materialize.foo",
1274    "fields": [
1275        {"name": "f1", "type": "int"},
1276        {"name": "f2", "type": "int"}
1277    ]
1278}
1279"#;
1280        let w = r#"
1281{
1282    "type": "record",
1283    "name": "com.materialize.foo",
1284    "fields": [
1285        {"name": "f1", "type": "int"}
1286    ]
1287}
1288"#;
1289        let r: Schema = r.parse().unwrap();
1290        let w: Schema = w.parse().unwrap();
1291        let err_str = if let Result::Err(AvroError::ResolveSchema(SchemaResolutionError(s))) =
1292            resolve_schemas(&w, &r)
1293        {
1294            s
1295        } else {
1296            panic!("Expected schema resolution failure");
1297        };
1298        assert_eq!(
1299            &err_str,
1300            "Reader field `com.materialize.foo.f2` not found in writer, and has no default"
1301        );
1302    }
1303
1304    #[mz_ore::test]
1305    fn test_duplicate_field_error() {
1306        let r = r#"
1307{
1308    "type": "record",
1309    "name": "com.materialize.bar",
1310    "fields": [
1311        {"name": "f1", "type": "int"},
1312        {"name": "f1", "type": "int"}
1313    ]
1314}
1315"#;
1316        let w = r#"
1317{
1318    "type": "record",
1319    "name": "com.materialize.bar",
1320    "fields": [
1321        {"name": "f1", "type": "int"}
1322    ]
1323}
1324"#;
1325        let r: Schema = r.parse().unwrap();
1326        let w: Schema = w.parse().unwrap();
1327        let err_str = if let Result::Err(AvroError::ResolveSchema(SchemaResolutionError(s))) =
1328            resolve_schemas(&w, &r)
1329        {
1330            s
1331        } else {
1332            panic!("Expected schema resolution failure");
1333        };
1334        assert_eq!(
1335            &err_str,
1336            "Duplicate field `com.materialize.bar.f1` in schema"
1337        );
1338    }
1339
1340    #[mz_ore::test]
1341    fn test_decimal_field_mismatch_error() {
1342        let r = r#"
1343{
1344    "type": "record",
1345    "name": "com.materialize.foo",
1346    "fields": [
1347        {"name": "f1", "type": {"type": "bytes", "logicalType": "decimal", "precision": 4, "scale": 2}}
1348    ]
1349}
1350"#;
1351        let w = r#"
1352{
1353    "type": "record",
1354    "name": "com.materialize.foo",
1355    "fields": [
1356        {"name": "f1", "type": {"type": "bytes", "logicalType": "decimal", "precision": 5, "scale": 1}}
1357    ]
1358}
1359"#;
1360        let r: Schema = r.parse().unwrap();
1361        let w: Schema = w.parse().unwrap();
1362        let err_str = if let Result::Err(AvroError::ResolveSchema(SchemaResolutionError(s))) =
1363            resolve_schemas(&w, &r)
1364        {
1365            s
1366        } else {
1367            panic!("Expected schema resolution failure");
1368        };
1369        assert_eq!(
1370            &err_str,
1371            "Decimal types must match in precision, scale, and fixed size. Got (5, 1, None); (4, 2. None) for field `com.materialize.foo.f1`"
1372        );
1373    }
1374}
mz_avro/reader.rs

mz_avro/
reader.rs
