This is an automated email from the ASF dual-hosted git repository.
milenkovicm 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 efaeda5ba feat(tui + rest): surface failed tasks (#1949)
efaeda5ba is described below
commit efaeda5ba689e14dc2b7368faf5438450dc1cfa4
Author: jgrim <[email protected]>
AuthorDate: Sat Jul 11 15:51:18 2026 +0200
feat(tui + rest): surface failed tasks (#1949)
* feat(rest): emit tasks with failure reason for Failed stages
* feat(tui): show task failure reason in the stage task view
* address review comments
---
ballista-cli/src/tui/domain/jobs/stages.rs | 14 ++++-
.../src/tui/ui/main/jobs/stage_tasks_popup.rs | 22 +++----
ballista/scheduler/src/api/handlers.rs | 73 +++++++++++++++++++++-
3 files changed, 91 insertions(+), 18 deletions(-)
diff --git a/ballista-cli/src/tui/domain/jobs/stages.rs
b/ballista-cli/src/tui/domain/jobs/stages.rs
index 3297aca43..5b81d0410 100644
--- a/ballista-cli/src/tui/domain/jobs/stages.rs
+++ b/ballista-cli/src/tui/domain/jobs/stages.rs
@@ -43,11 +43,19 @@ pub struct JobStageResponse {
pub tasks: Vec<Option<StageTaskResponse>>,
}
+// TaskStatus
+#[derive(Deserialize, Clone, Debug)]
+pub enum StageTaskStatus {
+ Running,
+ Successful,
+ Failed { reason: String },
+}
+
// TaskSummary
#[derive(Deserialize, Clone, Debug)]
pub struct StageTaskResponse {
pub id: usize,
- pub status: String,
+ pub status: StageTaskStatus,
pub partition_id: u32,
pub input_rows: usize,
pub output_rows: usize,
@@ -364,7 +372,7 @@ impl StagesGraph {
mod tests {
use super::{
JobStageResponse, JobStagesPopup, JobStagesResponse, StageTaskResponse,
- StagesGraph, TaskPercentiles,
+ StageTaskStatus, StagesGraph, TaskPercentiles,
};
fn make_percentiles() -> TaskPercentiles {
@@ -608,7 +616,7 @@ mod tests {
fn make_task(id: usize) -> StageTaskResponse {
StageTaskResponse {
id,
- status: "Completed".to_string(),
+ status: StageTaskStatus::Successful,
partition_id: id as u32,
input_rows: 0,
output_rows: 0,
diff --git a/ballista-cli/src/tui/ui/main/jobs/stage_tasks_popup.rs
b/ballista-cli/src/tui/ui/main/jobs/stage_tasks_popup.rs
index 76f9d769c..8c89a699b 100644
--- a/ballista-cli/src/tui/ui/main/jobs/stage_tasks_popup.rs
+++ b/ballista-cli/src/tui/ui/main/jobs/stage_tasks_popup.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::tui::app::App;
-use crate::tui::domain::jobs::stages::StageTaskResponse;
+use crate::tui::domain::jobs::stages::{StageTaskResponse, StageTaskStatus};
use crate::tui::ui::components::clear_area::clear_area;
use crate::tui::ui::vertical_scrollbar;
use ratatui::Frame;
@@ -101,21 +101,19 @@ fn build_stage_task_row(i: usize, task:
&StageTaskResponse, app: &App) -> Row<'s
app.theme.row_odd
};
- let status_style = match task.status.as_str() {
- "Running" => app.theme.status_running,
- "Queued" => app.theme.status_queued,
- "Successful" | "Completed" => app.theme.status_completed,
- "Failed" => app.theme.status_failed,
- _ => app.theme.status_unknown,
+ let status_text = match &task.status {
+ StageTaskStatus::Running =>
Text::from("Running").style(app.theme.status_running),
+ StageTaskStatus::Successful => {
+ Text::from("Successful").style(app.theme.status_completed)
+ }
+ StageTaskStatus::Failed { reason } => {
+ Text::from(reason.clone()).style(app.theme.status_failed)
+ }
};
Row::new(vec![
Cell::from(Text::from(task.id.to_string()).right_aligned()),
- Cell::from(
- Text::from(task.status.clone())
- .style(status_style)
- .centered(),
- ),
+ Cell::from(status_text.centered()),
Cell::from(Text::from(app.format_count(task.input_rows)).right_aligned()),
Cell::from(Text::from(app.format_count(task.output_rows)).right_aligned()),
Cell::from(Text::from(task.partition_id.to_string()).right_aligned()),
diff --git a/ballista/scheduler/src/api/handlers.rs
b/ballista/scheduler/src/api/handlers.rs
index ff886a69f..8da872253 100644
--- a/ballista/scheduler/src/api/handlers.rs
+++ b/ballista/scheduler/src/api/handlers.rs
@@ -22,9 +22,12 @@ use axum::{
extract::{Path, State},
response::{IntoResponse, Response},
};
+use ballista_core::serde::protobuf::failed_task::FailedReason::{
+ ExecutionError, ExecutorLost, FetchPartitionError, IoError, ResultLost,
TaskKilled,
+};
use ballista_core::serde::protobuf::job_status::Status;
use ballista_core::serde::protobuf::{
- ExecutorMetric, executor_metric::Metric, task_status,
+ ExecutorMetric, FailedTask, executor_metric::Metric, task_status,
};
use ballista_core::serde::scheduler::{
ExecutorOperatingSystemSpecification, ExecutorSpecification,
@@ -165,14 +168,17 @@ pub struct TaskSummary {
pub enum TaskStatus {
Running,
Successful,
- Failed,
+ Failed { reason: String, error: String },
}
impl From<&task_status::Status> for TaskStatus {
fn from(value: &task_status::Status) -> Self {
match value {
task_status::Status::Running(_) => TaskStatus::Running,
- task_status::Status::Failed(_) => TaskStatus::Failed,
+ task_status::Status::Failed(failed) => TaskStatus::Failed {
+ reason: failed_reason(failed),
+ error: failed.error.clone(),
+ },
task_status::Status::Successful(_) => TaskStatus::Successful,
}
}
@@ -637,6 +643,54 @@ pub async fn get_query_stages<
})
.collect();
}
+ ExecutionStage::Failed(failed_stage) => {
+ let metrics =
failed_stage.stage_metrics.as_deref().unwrap_or(&[]);
+ summary.stage_plan = Some(match plan_format {
+ PlanFormat::Default =>
displayable(failed_stage.plan.as_ref()).indent(false).to_string(),
+ PlanFormat::Tree =>
displayable(failed_stage.plan.as_ref()).tree_render().to_string(),
+ PlanFormat::Metrics =>
format_stage_metrics(failed_stage.plan.as_ref(), metrics),
+ });
+ summary.input_rows = get_combined_count(metrics,
"input_rows");
+ summary.output_rows = get_combined_count(metrics,
"output_rows");
+ summary.elapsed_compute = get_finished_stage_time(
+ &failed_stage
+ .task_infos
+ .iter()
+ .flatten()
+ .cloned()
+ .collect::<Vec<_>>(),
+ );
+
+ summary.tasks = failed_stage
+ .task_infos
+ .iter()
+ .enumerate()
+ .map(|(partition_id, task_info)| {
+ task_info.as_ref().map(|info| {
+ let (input_rows, output_rows) =
+ get_partition_counts(metrics,
partition_id);
+
+ let start_exec_time = info.start_exec_time
as u64;
+ let end_exec_time = info.end_exec_time as
u64;
+ let task_status: TaskStatus =
(&info.task_status).into();
+
+ TaskSummary {
+ id: info.task_id,
+ partition_id: partition_id as u32,
+ scheduled_time: info.scheduled_time as
u64,
+ launch_time: info.launch_time as u64,
+ start_exec_time,
+ end_exec_time,
+ exec_duration:
end_exec_time.saturating_sub(start_exec_time),
+ finish_time: info.finish_time as u64,
+ input_rows,
+ output_rows,
+ status: task_status,
+ }
+ })
+ })
+ .collect();
+ }
_ => {}
}
summary.task_duration_percentiles =
task_duration_percentiles(&summary.tasks);
@@ -761,6 +815,19 @@ fn get_running_stage_time(
}
}
+fn failed_reason(failed: &FailedTask) -> String {
+ match &failed.failed_reason {
+ Some(ExecutionError(_)) => "ExecutionError",
+ Some(FetchPartitionError(_)) => "FetchPartitionError",
+ Some(IoError(_)) => "IoError",
+ Some(ExecutorLost(_)) => "ExecutorLost",
+ Some(ResultLost(_)) => "ResultLost",
+ Some(TaskKilled(_)) => "TaskKilled",
+ None => "Failed",
+ }
+ .to_string()
+}
+
fn get_finished_stage_time(task_infos: &[TaskInfo]) -> Option<String> {
let min_start = task_infos
.iter()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
