This is an automated email from the ASF dual-hosted git repository.
Jefffrey pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 1c7f002e02 fix(arrow-avro): preserve object_store error source in the
async reader (#10496)
1c7f002e02 is described below
commit 1c7f002e02b69f317993621fc233013760b8413a
Author: ranflarion <[email protected]>
AuthorDate: Mon Aug 31 03:26:17 2026 -0400
fix(arrow-avro): preserve object_store error source in the async reader
(#10496)
# Which issue does this PR close?
- Closes #10493.
# Rationale for this change
`AvroObjectReader` converts every error from the underlying
`ObjectStore` with `AvroError::General(e.to_string())`, and the header
fetch loop wraps its error the same way, so the typed
`object_store::Error` is destroyed: `source()` returns nothing and no
downcast is possible. Callers implementing retry or error-classification
logic cannot distinguish a permanent `NotFound` from a transient
transport fault without string matching. `ParquetObjectReader` already
preserves the source via `E: Into<ParquetError>` in the identical
`spawn` helper plus `impl From<object_store::Error> for ParquetError`;
this aligns arrow-avro with that pattern.
# What changes are included in this PR?
- `impl From<object_store::Error> for AvroError` (feature-gated on
`object_store`), mapping to `AvroError::External(Box::new(e))`, which
already reports its inner error via `source()` and maps through
`ArrowError::from_external_error`.
- The private `spawn` helper bound changes from `E: Error` to `E:
Into<AvroError>`, mirroring `ParquetObjectReader::spawn`.
- The header fetch loop propagates the reader error instead of
reformatting it into `AvroError::General`.
# Are these changes tested?
Two new tests in `store.rs` assert that a read of a missing object
surfaces an error whose `source()` chain downcasts to
`object_store::Error::NotFound`, with and without a spawn runtime.
Existing suites pass unchanged.
# Are there any user-facing changes?
Error `Display` text changes for store failures (`External: ...` instead
of `Avro error: ...`). Variant selection within the `#[non_exhaustive]`
`AvroError` is not a stability contract, so this is not considered a
breaking API change.
---------
Co-authored-by: Jeffrey Vo <[email protected]>
---
arrow-avro/src/errors.rs | 7 ++++
arrow-avro/src/reader/async_reader/builder.rs | 9 +----
arrow-avro/src/reader/async_reader/store.rs | 57 ++++++++++++++++++++++++---
3 files changed, 59 insertions(+), 14 deletions(-)
diff --git a/arrow-avro/src/errors.rs b/arrow-avro/src/errors.rs
index dd102abac7..e9a8ca3b87 100644
--- a/arrow-avro/src/errors.rs
+++ b/arrow-avro/src/errors.rs
@@ -135,6 +135,13 @@ impl From<ArrowError> for AvroError {
}
}
+#[cfg(feature = "object_store")]
+impl From<object_store::Error> for AvroError {
+ fn from(e: object_store::Error) -> AvroError {
+ AvroError::External(Box::new(e))
+ }
+}
+
impl From<AvroError> for io::Error {
fn from(e: AvroError) -> Self {
io::Error::other(e)
diff --git a/arrow-avro/src/reader/async_reader/builder.rs
b/arrow-avro/src/reader/async_reader/builder.rs
index d3cca70425..064adfad3e 100644
--- a/arrow-avro/src/reader/async_reader/builder.rs
+++ b/arrow-avro/src/reader/async_reader/builder.rs
@@ -154,14 +154,7 @@ where
break;
}
- let current_data = reader
- .get_bytes(range_to_fetch.clone())
- .await
- .map_err(|err| {
- AvroError::General(format!(
- "Error fetching Avro header from file reader: {err}"
- ))
- })?;
+ let current_data = reader.get_bytes(range_to_fetch.clone()).await?;
if current_data.is_empty() {
return Err(AvroError::EOF(
"Unexpected EOF while fetching header data".into(),
diff --git a/arrow-avro/src/reader/async_reader/store.rs
b/arrow-avro/src/reader/async_reader/store.rs
index 44f0b3b42b..d1fdb3e15e 100644
--- a/arrow-avro/src/reader/async_reader/store.rs
+++ b/arrow-avro/src/reader/async_reader/store.rs
@@ -23,7 +23,6 @@ use futures::{FutureExt, TryFutureExt};
use object_store::ObjectStore;
use object_store::ObjectStoreExt;
use object_store::path::Path;
-use std::error::Error;
use std::ops::Range;
use std::sync::Arc;
use tokio::runtime::Handle;
@@ -76,7 +75,7 @@ impl AvroObjectReader {
+ Send
+ 'static,
O: Send + 'static,
- E: Error + Send + 'static,
+ E: Into<AvroError> + Send + 'static,
{
match &self.runtime {
Some(handle) => {
@@ -89,13 +88,11 @@ impl AvroObjectReader {
Err(e) => Err(AvroError::External(Box::new(e))),
Ok(p) => std::panic::resume_unwind(p),
},
- |res| res.map_err(|e|
AvroError::General(e.to_string())),
+ |res| res.map_err(Into::into),
)
.boxed()
}
- None => f(&self.store, &self.path)
- .map_err(|e| AvroError::General(e.to_string()))
- .boxed(),
+ None => f(&self.store, &self.path).map_err(Into::into).boxed(),
}
}
}
@@ -116,3 +113,51 @@ impl AsyncFileReader for AvroObjectReader {
self.spawn(|store, path| async move { store.get_ranges(path,
&ranges).await }.boxed())
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use object_store::memory::InMemory;
+
+ fn find_object_store_error(err: &AvroError) ->
Option<&object_store::Error> {
+ let mut source: Option<&(dyn std::error::Error + 'static)> = Some(err);
+ while let Some(e) = source {
+ if let Some(os) = e.downcast_ref::<object_store::Error>() {
+ return Some(os);
+ }
+ source = e.source();
+ }
+ None
+ }
+
+ #[tokio::test]
+ async fn test_get_bytes_preserves_object_store_error_source() {
+ let store = Arc::new(InMemory::new());
+ #[expect(deprecated)]
+ let mut reader = AvroObjectReader::new(store,
Path::from("missing.avro"));
+ let err = reader.get_bytes(0..10).await.unwrap_err();
+ assert!(
+ matches!(
+ find_object_store_error(&err),
+ Some(object_store::Error::NotFound { .. })
+ ),
+ "expected NotFound in source chain, got: {err}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_get_bytes_on_runtime_preserves_object_store_error_source() {
+ let store = Arc::new(InMemory::new());
+ #[expect(deprecated)]
+ let mut reader = AvroObjectReader::new(store,
Path::from("missing.avro"))
+ .with_runtime(Handle::current());
+ let err = reader.get_bytes(0..10).await.unwrap_err();
+ assert!(
+ matches!(
+ find_object_store_error(&err),
+ Some(object_store::Error::NotFound { .. })
+ ),
+ "expected NotFound in source chain, got: {err}"
+ );
+ }
+}
