villebro opened a new pull request, #2276: URL: https://github.com/apache/datafusion-ballista/pull/2276
<!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> Follow-up to #2202 (review discussion [r3737946864](https://github.com/apache/datafusion-ballista/pull/2202#discussion_r3737946864)). # Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> #2202 kept shuffle-fetch failures structural until task-failure classification, but carried the structured `BallistaError` across DataFusion as `ArrowError::ExternalError(Box<BallistaError>)`. Because DataFusion's `RecordBatchStream` yields `Result<RecordBatch, DataFusionError>`, that Arrow error auto-converts (via `From<ArrowError> for DataFusionError`) into `DataFusionError::ArrowError(ArrowError::ExternalError(..))` — a double wrap. Recovery then had to recurse through both the DataFusion and Arrow layers (`find_fetch_failed` → `fetch_failed_in_datafusion` → `fetch_failed_in_arrow`) to find the original error. `DataFusionError::External` is the purpose-built variant for errors originating outside DataFusion. It is a single wrap, and `find_root()` already traverses `Shared` / `Context` / `Diagnostic` and lands on it. Using it removes the Arrow intermediate and collapses recovery to one downcast. This is an internal refactor: wrapped errors are classified into the `FailedTask` proto entirely inside the executor process before crossing any wire, so the wrapping convention can be changed wholesale without affecting the protocol. # What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> A structured `BallistaError` (e.g. a shuffle `FetchFailed`) has to travel from the shuffle stream, through DataFusion operators that re-wrap errors, back to task-failure classification. Before, it was carried through Arrow, which DataFusion then re-wrapped — a double layer that recovery had to unwind through both the DataFusion and Arrow sides: ``` carry: DataFusionError::ArrowError(ArrowError::ExternalError(Box<BallistaError>)) recover: find_root() -> ArrowError -> ExternalError -> downcast::<BallistaError> ``` After, it is carried as `External`, the variant meant for foreign errors, which `find_root()` already sees through — so recovery is a single downcast: ``` carry: DataFusionError::External(Box<BallistaError>) recover: find_root() -> External -> downcast::<BallistaError> ``` Concretely: - `BallistaError::into_datafusion` is the single helper for the `carry` step, used at the shuffle-reader fetch-stream boundary and the shuffle-writer drain handoff. - `find_fetch_failed` / `fetch_failed_in_arrow` / `fetch_failed_in_datafusion` collapse into one `find_root()` + `External` downcast, and `is_retryable_io` sees through the same wrapper so IO errors stay retryable. - Sites that already held a `DataFusionError` (`summaries_to_batch` in both writers, the `execute_query_pull` / `execute_query_push` client result streams) no longer wrap it in Arrow just to have it flattened back — the redundant round-trip is dropped. - The `distributed_query` fetch path, which does carry a `BallistaError`, now routes through the same `into_datafusion` helper instead of hand-building `DataFusionError::External`. - The error-classification tests are consolidated into property-based tests (e.g. "IO stays retryable through any wrapper") that exercise the bare, `External`, `Shared`, and `Context` shapes in one place instead of a separate test per shape. Failure classification is behavior-preserving: the fetch-partition, retryable-IO, task-killed, and execution-error reasons the scheduler acts on are unchanged, including when the error arrives under `Shared` / `Context` layers. # Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> No API or documentation changes. Free-text task error messages lose the redundant wrapper layer: the `Arrow error: External error:` prefix collapses to `External error:`, and the two IO-error messages (`Task failed due to Ballista IO error` / `Task failed due to DataFusion IO error`) are unified to `Task failed due to IO error`. Structured failure classification — the fetch-partition, retryable-IO, task-killed, and execution-error reasons the scheduler acts on — is unchanged. No breaking changes to public APIs. <!-- If there are any breaking changes to public APIs, please add the `api change` label. --> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
