AvroError enum for arrow-avro crate

arrow-avro/src/compression.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::errors::AvroError;
 use arrow_schema::ArrowError;
 #[cfg(any(
     feature = "deflate",
@@ -47,7 +48,7 @@ pub enum CompressionCodec {
 
 impl CompressionCodec {
     #[allow(unused_variables)]
-    pub(crate) fn decompress(&self, block: &[u8]) -> Result<Vec<u8>, ArrowError> {
+    pub(crate) fn decompress(&self, block: &[u8]) -> Result<Vec<u8>, AvroError> {
         match self {
             #[cfg(feature = "deflate")]
             CompressionCodec::Deflate => {
@@ -57,7 +58,7 @@ impl CompressionCodec {
                 Ok(out)
             }
             #[cfg(not(feature = "deflate"))]
-            CompressionCodec::Deflate => Err(ArrowError::ParseError(
+            CompressionCodec::Deflate => Err(AvroError::ParseError(
                 "Deflate codec requires deflate feature".to_string(),
             )),
             #[cfg(feature = "snappy")]
@@ -70,50 +71,56 @@ impl CompressionCodec {
                 let mut decoder = snap::raw::Decoder::new();
                 let decoded = decoder
                     .decompress_vec(block)
-                    .map_err(|e| ArrowError::ExternalError(Box::new(e)))?;
+                    .map_err(|e| AvroError::External(Box::new(e)))?;
 
                 let checksum = crc::Crc::<u32>::new(&crc::CRC_32_ISO_HDLC).checksum(&decoded);
                 if checksum != u32::from_be_bytes(crc.try_into().unwrap()) {
-                    return Err(ArrowError::ParseError("Snappy CRC mismatch".to_string()));
+                    return Err(AvroError::ParseError("Snappy CRC mismatch".to_string()));
                 }
                 Ok(decoded)
             }
             #[cfg(not(feature = "snappy"))]
-            CompressionCodec::Snappy => Err(ArrowError::ParseError(
+            CompressionCodec::Snappy => Err(AvroError::ParseError(
                 "Snappy codec requires snappy feature".to_string(),
             )),
 
             #[cfg(feature = "zstd")]
             CompressionCodec::ZStandard => {
                 let mut decoder = zstd::Decoder::new(block)?;
                 let mut out = Vec::new();
-                decoder.read_to_end(&mut out)?;
+                decoder
+                    .read_to_end(&mut out)
+                    .map_err(|e| AvroError::External(Box::new(e)))?;
                 Ok(out)
             }
             #[cfg(not(feature = "zstd"))]
-            CompressionCodec::ZStandard => Err(ArrowError::ParseError(
+            CompressionCodec::ZStandard => Err(AvroError::ParseError(
                 "ZStandard codec requires zstd feature".to_string(),
             )),
             #[cfg(feature = "bzip2")]
             CompressionCodec::Bzip2 => {
                 let mut decoder = bzip2::read::BzDecoder::new(block);
                 let mut out = Vec::new();
-                decoder.read_to_end(&mut out)?;
+                decoder
+                    .read_to_end(&mut out)
+                    .map_err(|e| AvroError::External(Box::new(e)))?;
                 Ok(out)
             }
             #[cfg(not(feature = "bzip2"))]
-            CompressionCodec::Bzip2 => Err(ArrowError::ParseError(
+            CompressionCodec::Bzip2 => Err(AvroError::ParseError(
                 "Bzip2 codec requires bzip2 feature".to_string(),
             )),
             #[cfg(feature = "xz")]
             CompressionCodec::Xz => {
                 let mut decoder = xz::read::XzDecoder::new(block);
                 let mut out = Vec::new();
-                decoder.read_to_end(&mut out)?;
+                decoder
+                    .read_to_end(&mut out)
+                    .map_err(|e| AvroError::External(Box::new(e)))?;
                 Ok(out)
             }
             #[cfg(not(feature = "xz"))]
-            CompressionCodec::Xz => Err(ArrowError::ParseError(
+            CompressionCodec::Xz => Err(AvroError::ParseError(
                 "XZ codec requires xz feature".to_string(),
             )),
         }

arrow-avro/src/errors.rs

@@ -0,0 +1,148 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Common Avro errors and macros.
+
+use arrow_schema::ArrowError;
+use core::num::TryFromIntError;
+use std::error::Error;
+use std::string::FromUtf8Error;
+use std::{io, str};
+
+/// Avro error enumeration
+
+#[derive(Debug)]
+#[non_exhaustive]
+pub enum AvroError {
+    /// General Avro error.
+    /// Returned when code violates normal workflow of working with Avro data.
+    General(String),
+    /// "Not yet implemented" Avro error.
+    /// Returned when functionality is not yet available.
+    NYI(String),
+    /// "End of file" Avro error.
+    /// Returned when IO related failures occur, e.g. when there are not enough bytes to
+    /// decode.
+    EOF(String),
+    /// Arrow error.
+    /// Returned when reading into arrow or writing from arrow.
+    ArrowError(Box<ArrowError>),
+    /// Error when the requested index is more than the
+    /// number of items expected
+    IndexOutOfBound(usize, usize),
+    /// Error indicating that an unexpected or bad argument was passed to a function.
+    InvalidArgument(String),
+    /// Error indicating that a value could not be parsed.
+    ParseError(String),
+    /// Error indicating that a schema is invalid.
+    SchemaError(String),
+    /// An external error variant
+    External(Box<dyn Error + Send + Sync>),
+    /// Error during IO operations
+    IoError(String, io::Error),
+    /// Returned when a function needs more data to complete properly. The `usize` field indicates
+    /// the total number of bytes required, not the number of additional bytes.
+    NeedMoreData(usize),
+    /// Returned when a function needs more data to complete properly.
+    /// The `Range<u64>` indicates the range of bytes that are needed.
+    NeedMoreDataRange(std::ops::Range<u64>),
+}
+
+impl std::fmt::Display for AvroError {
+    fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match &self {
+            AvroError::General(message) => {
+                write!(fmt, "Avro error: {message}")
+            }
+            AvroError::NYI(message) => write!(fmt, "NYI: {message}"),
+            AvroError::EOF(message) => write!(fmt, "EOF: {message}"),
+            AvroError::ArrowError(message) => write!(fmt, "Arrow: {message}"),
+            AvroError::IndexOutOfBound(index, bound) => {
+                write!(fmt, "Index {index} out of bound: {bound}")
+            }
+            AvroError::InvalidArgument(message) => {
+                write!(fmt, "Invalid argument: {message}")
+            }
+            AvroError::ParseError(message) => write!(fmt, "Parser error: {message}"),
+            AvroError::SchemaError(message) => write!(fmt, "Schema error: {message}"),
+            AvroError::External(e) => write!(fmt, "External: {e}"),
+            AvroError::IoError(message, e) => write!(fmt, "I/O Error: {message}: {e}"),
+            AvroError::NeedMoreData(needed) => write!(fmt, "NeedMoreData: {needed}"),
+            AvroError::NeedMoreDataRange(range) => {
+                write!(fmt, "NeedMoreDataRange: {}..{}", range.start, range.end)
+            }
+        }
+    }
+}
+
+impl Error for AvroError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            AvroError::External(e) => Some(e.as_ref()),
+            AvroError::ArrowError(e) => Some(e.as_ref()),
+            AvroError::IoError(_, e) => Some(e),
+            _ => None,
+        }
+    }
+}
+
+impl From<TryFromIntError> for AvroError {
+    fn from(e: TryFromIntError) -> AvroError {
+        AvroError::General(format!("Integer overflow: {e}"))
+    }
+}
+
+impl From<io::Error> for AvroError {
+    fn from(e: io::Error) -> AvroError {
+        AvroError::External(Box::new(e))
+    }
+}
+
+impl From<str::Utf8Error> for AvroError {
+    fn from(e: str::Utf8Error) -> AvroError {
+        AvroError::External(Box::new(e))
+    }
+}
+
+impl From<FromUtf8Error> for AvroError {
+    fn from(e: FromUtf8Error) -> AvroError {
+        AvroError::External(Box::new(e))
+    }
+}
+
+impl From<ArrowError> for AvroError {
+    fn from(e: ArrowError) -> Self {
+        AvroError::ArrowError(Box::new(e))
+    }
+}
+
+impl From<AvroError> for io::Error {
+    fn from(e: AvroError) -> Self {
+        io::Error::other(e)
+    }
+}
+
+impl From<AvroError> for ArrowError {
+    fn from(e: AvroError) -> Self {
+        match e {
+            AvroError::External(inner) => ArrowError::from_external_error(inner),
+            AvroError::IoError(msg, err) => ArrowError::IoError(msg, err),
+            AvroError::ArrowError(inner) => *inner,
+            other => ArrowError::AvroError(other.to_string()),
+        }
+    }
+}

arrow-avro/src/lib.rs

@@ -195,6 +195,9 @@ pub mod compression;
 /// Avro data types and Arrow data types.
 pub mod codec;
 
+/// AvroError variants
+pub mod errors;
+
 /// Extension trait for AvroField to add Utf8View support
 ///
 /// This trait adds methods for working with Utf8View support to the AvroField struct.

arrow-avro/src/reader/block.rs

@@ -17,8 +17,8 @@
 
 //! Decoder for [`Block`]
 
+use crate::errors::AvroError;
 use crate::reader::vlq::VLQDecoder;
-use arrow_schema::ArrowError;
 
 /// A file data block
 ///
@@ -75,14 +75,14 @@ impl BlockDecoder {
     /// can then be used again to read the next block, if any
     ///
     /// [`BufRead::fill_buf`]: std::io::BufRead::fill_buf
-    pub fn decode(&mut self, mut buf: &[u8]) -> Result<usize, ArrowError> {
+    pub fn decode(&mut self, mut buf: &[u8]) -> Result<usize, AvroError> {
         let max_read = buf.len();
         while !buf.is_empty() {
             match self.state {
                 BlockDecoderState::Count => {
                     if let Some(c) = self.vlq_decoder.long(&mut buf) {
                         self.in_progress.count = c.try_into().map_err(|_| {
-                            ArrowError::ParseError(format!(
+                            AvroError::ParseError(format!(
                                 "Block count cannot be negative, got {c}"
                             ))
                         })?;
@@ -93,9 +93,7 @@ impl BlockDecoder {
                 BlockDecoderState::Size => {
                     if let Some(c) = self.vlq_decoder.long(&mut buf) {
                         self.bytes_remaining = c.try_into().map_err(|_| {
-                            ArrowError::ParseError(format!(
-                                "Block size cannot be negative, got {c}"
-                            ))
+                            AvroError::ParseError(format!("Block size cannot be negative, got {c}"))
                         })?;
 
                         self.in_progress.data.reserve(self.bytes_remaining);