This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new b16270801 see through Shared wrapper when classifying task IO errors
(#2119)
b16270801 is described below
commit b16270801249a3422e9384ddf4a52bf92e332e87
Author: goingforstudying-ctrl <[email protected]>
AuthorDate: Tue Jul 21 08:46:13 2026 -0400
see through Shared wrapper when classifying task IO errors (#2119)
A retryable IO error raised on a join's shared build side arrives at the
FailedTask classifier wrapped in DataFusionError::Shared, so the shallow
matches! on the outermost variant never sees the IoError inside and the
task is marked non-retryable. Classify on find_root() instead so the
retryability decision is based on the root error regardless of wrapping.
Co-authored-by: goingforstudying-ctrl
<[email protected]>
---
ballista/core/src/error.rs | 75 +++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 74 insertions(+), 1 deletion(-)
diff --git a/ballista/core/src/error.rs b/ballista/core/src/error.rs
index b0ce594f3..b5a7a82b0 100644
--- a/ballista/core/src/error.rs
+++ b/ballista/core/src/error.rs
@@ -235,7 +235,7 @@ impl From<BallistaError> for FailedTask {
}
}
BallistaError::DataFusionError(e)
- if matches!(*e, DataFusionError::IoError(_)) =>
+ if matches!(e.find_root(), DataFusionError::IoError(_)) =>
{
FailedTask {
error: format!("Task failed due to DataFusion IO error:
{e:?}"),
@@ -256,3 +256,76 @@ impl From<BallistaError> for FailedTask {
}
impl Error for BallistaError {}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::sync::Arc;
+
+ fn io_failed_task(e: BallistaError) -> FailedTask {
+ FailedTask::from(e)
+ }
+
+ #[test]
+ fn bare_datafusion_io_error_is_retryable() {
+ let e =
BallistaError::DataFusionError(Box::new(DataFusionError::IoError(
+ io::Error::new(io::ErrorKind::ConnectionReset, "connection reset"),
+ )));
+ let task = io_failed_task(e);
+ assert!(task.retryable);
+ assert!(matches!(task.failed_reason, Some(FailedReason::IoError(_))));
+ }
+
+ #[test]
+ fn shared_wrapped_io_error_is_retryable() {
+ // Errors from a join's shared build side arrive as
Shared(Arc<IoError>);
+ // the classifier must see through the wrapper or it will not retry a
+ // transient IO failure.
+ let inner = DataFusionError::IoError(io::Error::new(
+ io::ErrorKind::ConnectionReset,
+ "connection reset",
+ ));
+ let shared = DataFusionError::Shared(Arc::new(inner));
+ let e = BallistaError::DataFusionError(Box::new(shared));
+ let task = io_failed_task(e);
+ assert!(task.retryable);
+ assert!(matches!(task.failed_reason, Some(FailedReason::IoError(_))));
+ }
+
+ #[test]
+ fn context_wrapped_shared_io_error_is_retryable() {
+ let inner = DataFusionError::IoError(io::Error::other("s3 timeout"));
+ let shared = DataFusionError::Shared(Arc::new(inner));
+ let ctx = shared.context("reading join build side");
+ let e = BallistaError::DataFusionError(Box::new(ctx));
+ let task = io_failed_task(e);
+ assert!(task.retryable);
+ assert!(matches!(task.failed_reason, Some(FailedReason::IoError(_))));
+ }
+
+ #[test]
+ fn non_io_datafusion_error_stays_non_retryable() {
+ let e = BallistaError::DataFusionError(Box::new(DataFusionError::Plan(
+ "bad plan".to_string(),
+ )));
+ let task = io_failed_task(e);
+ assert!(!task.retryable);
+ assert!(matches!(
+ task.failed_reason,
+ Some(FailedReason::ExecutionError(_))
+ ));
+ }
+
+ #[test]
+ fn shared_wrapped_non_io_error_stays_non_retryable() {
+ let inner = DataFusionError::Plan("bad plan".to_string());
+ let shared = DataFusionError::Shared(Arc::new(inner));
+ let e = BallistaError::DataFusionError(Box::new(shared));
+ let task = io_failed_task(e);
+ assert!(!task.retryable);
+ assert!(matches!(
+ task.failed_reason,
+ Some(FailedReason::ExecutionError(_))
+ ));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
