You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by mi...@apache.org on 2026/08/18 20:39:50 UTC
(datafusion-ballista) branch main updated: fix(tui): use the shared API wire types instead of local copies (#2258)
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 a12fc9d60 fix(tui): use the shared API wire types instead of local copies (#2258)
a12fc9d60 is described below
commit a12fc9d6059699ba413d7be4f6fa18cca2ffa0b1
Author: Andy Grove <ag...@apache.org>
AuthorDate: Tue Aug 18 14:39:45 2026 -0600
fix(tui): use the shared API wire types instead of local copies (#2258)
* refactor(scheduler): extract REST DTOs into ballista-history and factor out dto_build
Move the scheduler's REST response types into a new leaf crate,
`ballista-history`, and pull the graph-to-DTO construction out of the
axum handlers into a pure `api::dto_build` module.
Behavior preserving: the same DTOs are produced from the same state, so
live REST responses are byte-identical. The existing handler tests cover
this, and the helper unit tests move alongside the functions they test.
This is the first step toward a history server that replays completed
jobs and serves the same `/api/*` responses without a live scheduler.
Splitting the DTOs into a serde-only crate lets that server build the
identical wire types without depending on the scheduler's live
execution graph, and moving construction out of the handlers means it
can run against state that did not come from a handler request.
`JobResponse::job_id` becomes a `String` rather than `ballista_core::JobId`
so the new crate stays serde-only. `JobId` is `#[serde(transparent)]`
over `String`, so the JSON is unchanged.
* refactor(scheduler): tighten the DTO extraction
Follow-up cleanups on the extraction:
- Collapse the three near-identical ExecutionStage arms in
graph_to_query_stages into one destructuring match, dropping the
mutable placeholder-zero summary.
- Take PlanFormat by value instead of &JobQueryParams, and move
PlanFormat into ballista-history. It is part of the wire contract, and
the pure builder no longer imports an axum query-param type back out of
the handler module.
- Inject `now` into graph_to_query_stages rather than reading the clock,
so replaying a stored log renders stable elapsed times.
- Share percent_complete and min_start_time; use displayable() instead of
the longhand DisplayableExecutionPlan::new().
- Drop the dead JobConfig alias and the unused serde_json dev-dependency,
make task_status_to_dto private, and remove a duplicated test.
- Enable #![warn(missing_docs)] on ballista-history and document the
types, matching the other Ballista crates.
- Register ballista-history with the release tooling: version bump
script, publish order, and crate dependency graph.
* docs(history): state the write-once, replay-verbatim data flow
The previous wording left it ambiguous whether the history server
re-derives responses from stored execution state or replays stored DTOs.
It replays them: the scheduler builds each response once against the live
graph and writes it to the event log, so byte-identical output is a
structural property rather than two implementations agreeing.
Also records the consequence, that anything not captured at write time
cannot be recovered at replay time.
* refactor: rename ballista-history to ballista-api-types
The crate holds the /api/* wire types, and it has three parties, not one:
the scheduler serves them, the web TUI deserializes them, and a future
history server will serve replayed copies. Naming it after the history
server made it awkward for the TUI, which parses live scheduler responses
and today keeps its own duplicate declarations.
Renaming it after the contract it defines removes that friction. The
event-log schema, writer, and reader can then land as a separate
ballista-history crate that depends on this one.
* fix(tui): use the shared API wire types instead of local copies
The TUI declared its own structs for the scheduler's /api/* responses.
Nothing kept them in step, so 2ed3464df changed TaskSummary::partition_id
from u32 to Vec<u32> without touching the TUI, and the stages popup has
been unable to parse a stage response since.
Replace the local declarations with the shared ballista-api-types
definitions, so the next scheduler-side field change is a compile error
here rather than a runtime parse failure:
Job -> JobResponse
JobStagesResponse -> QueryStagesResponse
JobStageResponse -> QueryStageSummary
StageTaskResponse -> TaskSummary
StageTaskStatus -> TaskStatus
TaskPercentiles -> Percentiles
Job's four status predicates move to a JobStatusExt extension trait,
since the type now lives in another crate.
Two display sites change as a result. Multi-partition tasks render the
whole partition list rather than a single id, and a failed task can now
show the scheduler's error text alongside the reason.
Closes #2257.
* Update ballista-cli/src/tui/ui/main/jobs/mod.rs
Co-authored-by: Martin Grigorov <ma...@users.noreply.github.com>
* Make INVALID_DATE pub(crate)
* Declare ballista-api-types in the workspace dependencies
* Print the error too for failed jobs, not just the category
---------
Co-authored-by: Martin Grigorov <ma...@users.noreply.github.com>
Co-authored-by: Martin Tzvetanov Grigorov <mg...@apache.org>
---
Cargo.lock | 1 +
Cargo.toml | 1 +
ballista-cli/Cargo.toml | 3 +
ballista-cli/src/tui/app.rs | 6 +-
ballista-cli/src/tui/domain/jobs.rs | 47 +++++---
ballista-cli/src/tui/domain/jobs/stages.rs | 132 +++++++++++----------
ballista-cli/src/tui/mod.rs | 12 +-
.../src/tui/ui/main/jobs/job_stages_popup.rs | 6 +-
ballista-cli/src/tui/ui/main/jobs/mod.rs | 15 ++-
.../src/tui/ui/main/jobs/stage_plan_popup.rs | 15 ++-
.../src/tui/ui/main/jobs/stage_tasks_popup.rs | 18 ++-
ballista/history/Cargo.toml | 2 +-
ballista/scheduler/Cargo.toml | 2 +-
13 files changed, 151 insertions(+), 109 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 94414f74d..366679d59 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1079,6 +1079,7 @@ name = "ballista-cli"
version = "54.0.0"
dependencies = [
"ballista",
+ "ballista-api-types",
"chrono",
"clap 4.6.6",
"config",
diff --git a/Cargo.toml b/Cargo.toml
index d989a3b4a..1c17dbe18 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -39,6 +39,7 @@ rust-version = "1.94.0"
[workspace.dependencies]
arrow = { version = "59.2.0", features = ["ipc_compression"] }
arrow-flight = { version = "59.2.0", features = ["flight-sql-experimental"] }
+ballista-api-types = { path = "ballista/api-types", version = "54.0.0" }
clap = { version = "4.5", features = ["derive", "cargo"] }
# DataFusion crates are pinned to the 55.0.0-rc3 tag via `[patch.crates-io]`
diff --git a/ballista-cli/Cargo.toml b/ballista-cli/Cargo.toml
index a61a63247..f120dfa2f 100644
--- a/ballista-cli/Cargo.toml
+++ b/ballista-cli/Cargo.toml
@@ -40,6 +40,7 @@ mimalloc = { workspace = true, optional = true }
rustyline = { version = "18.0.0", optional = true }
# TUI/web shared deps
+ballista-api-types = { workspace = true, optional = true }
chrono = { version = "0.4", default-features = false, features = ["std", "clock"], optional = true }
clap = { workspace = true, features = ["derive", "cargo"] }
config = { version = "0.15.23", default-features = false, features = ["yaml"], optional = true }
@@ -89,6 +90,7 @@ cli = [
]
tui = [
+ "dep:ballista-api-types",
"dep:chrono",
"dep:config",
"dep:crossterm",
@@ -111,6 +113,7 @@ tui = [
]
web = [
+ "dep:ballista-api-types",
"dep:chrono",
"dep:config",
"dep:critical-section",
diff --git a/ballista-cli/src/tui/app.rs b/ballista-cli/src/tui/app.rs
index 8dc7a5165..abc08313d 100644
--- a/ballista-cli/src/tui/app.rs
+++ b/ballista-cli/src/tui/app.rs
@@ -18,6 +18,7 @@
#[cfg(not(feature = "web"))]
use crate::tui::TuiError;
use crate::tui::TuiResult;
+use crate::tui::domain::jobs::JobStatusExt;
use crate::tui::event::Event;
#[cfg(feature = "web")]
use crate::tui::event::web::Sender;
@@ -57,7 +58,7 @@ use crate::tui::ui::{
load_metrics_data, load_stage_plan,
};
-const INVALID_DATE: &str = "Invalid date";
+pub(crate) const INVALID_DATE: &str = "Invalid date";
#[derive(Debug, PartialEq)]
enum Views {
@@ -1335,6 +1336,9 @@ mod tests {
num_stages: 1,
completed_stages: 0,
percent_complete: 0,
+ logical_plan: None,
+ physical_plan: None,
+ stage_plan: None,
}
}
diff --git a/ballista-cli/src/tui/domain/jobs.rs b/ballista-cli/src/tui/domain/jobs.rs
index eb2266370..3f7432158 100644
--- a/ballista-cli/src/tui/domain/jobs.rs
+++ b/ballista-cli/src/tui/domain/jobs.rs
@@ -21,33 +21,38 @@ use ratatui::widgets::{ScrollbarState, TableState};
use serde::Deserialize;
use std::collections::BTreeMap;
-#[derive(Deserialize, Clone, Debug)]
-pub struct Job {
- pub job_id: String,
- pub job_name: String,
- pub status: String, // Running, Completed, Failed, Canceled
- pub job_status: String, // human-readable status/failure detail, e.g. "Failed: <reason>"
- pub start_time: i64,
- pub end_time: i64,
- pub num_stages: usize,
- pub completed_stages: usize,
- pub percent_complete: u8,
+// Shared with the scheduler that serves it, so a field change there is a
+// compile error here rather than a runtime deserialization failure (#2257).
+pub use ballista_api_types::dto::JobResponse as Job;
+
+/// Status predicates over the shared [`Job`] type. An extension trait because
+/// `Job` is defined in `ballista-api-types`, which is deliberately free of any
+/// TUI concerns.
+pub trait JobStatusExt {
+ /// Job is accepted but not yet scheduled.
+ fn is_queued(&self) -> bool;
+ /// Job is currently executing.
+ fn is_running(&self) -> bool;
+ /// Job finished successfully.
+ fn is_completed(&self) -> bool;
+ /// Job terminated with an error.
+ fn is_failed(&self) -> bool;
}
-impl Job {
- pub fn is_queued(&self) -> bool {
+impl JobStatusExt for Job {
+ fn is_queued(&self) -> bool {
self.status == "Queued"
}
- pub fn is_running(&self) -> bool {
+ fn is_running(&self) -> bool {
self.status == "Running"
}
- pub fn is_completed(&self) -> bool {
+ fn is_completed(&self) -> bool {
self.status == "Completed"
}
- pub fn is_failed(&self) -> bool {
+ fn is_failed(&self) -> bool {
self.status == "Failed"
}
}
@@ -454,8 +459,8 @@ mod tests {
id: &str,
name: &str,
status: &str,
- start_time: i64,
- end_time: i64,
+ start_time: u64,
+ end_time: u64,
num_stages: usize,
completed_stages: usize,
percent_complete: u8,
@@ -470,6 +475,9 @@ mod tests {
num_stages,
completed_stages,
percent_complete,
+ logical_plan: None,
+ physical_plan: None,
+ stage_plan: None,
}
}
@@ -748,6 +756,9 @@ mod tests {
num_stages: 1,
completed_stages: 0,
percent_complete: 0,
+ logical_plan: None,
+ physical_plan: None,
+ stage_plan: None,
};
assert_eq!(job.status, "Failed");
assert_eq!(job.job_status, "Failed: division by zero");
diff --git a/ballista-cli/src/tui/domain/jobs/stages.rs b/ballista-cli/src/tui/domain/jobs/stages.rs
index 5b81d0410..215311fe4 100644
--- a/ballista-cli/src/tui/domain/jobs/stages.rs
+++ b/ballista-cli/src/tui/domain/jobs/stages.rs
@@ -16,65 +16,16 @@
// under the License.
use ratatui::widgets::{ScrollbarState, TableState};
-use serde::Deserialize;
use std::fmt::Display;
-#[derive(Deserialize, Clone, Debug)]
-pub struct JobStagesResponse {
- pub stages: Vec<JobStageResponse>,
-}
-
-#[derive(Deserialize, Clone, Debug)]
-pub struct JobStageResponse {
- #[serde(rename = "stage_id")]
- pub id: String,
- #[serde(rename = "stage_status")]
- pub status: String,
- #[serde(rename = "stage_plan", default)]
- pub plan: String,
- pub input_rows: usize,
- pub output_rows: usize,
- pub elapsed_compute: Option<String>,
- #[serde(default)]
- pub task_duration_percentiles: Option<TaskPercentiles>,
- #[serde(default)]
- pub task_input_percentiles: Option<TaskPercentiles>,
- #[serde(default)]
- 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: StageTaskStatus,
- pub partition_id: u32,
- pub input_rows: usize,
- pub output_rows: usize,
- pub scheduled_time: u64,
- pub launch_time: u64,
- pub start_exec_time: u64,
- pub end_exec_time: u64,
- pub finish_time: u64,
-}
-
-// Percentiles
-#[derive(Deserialize, Clone, Debug)]
-pub struct TaskPercentiles {
- pub min: u64,
- pub max: u64,
- pub median: u64,
- pub p25: u64,
- pub p75: u64,
-}
+// The `/api/*` response shapes are defined once in `ballista-api-types` and
+// shared with the scheduler that serves them. Re-exported here under the names
+// the TUI already used, so a scheduler-side field change is a compile error
+// rather than a deserialization failure at runtime (#2257).
+pub use ballista_api_types::dto::{
+ QueryStageSummary as JobStageResponse, QueryStagesResponse as JobStagesResponse,
+ TaskStatus as StageTaskStatus, TaskSummary as StageTaskResponse,
+};
#[derive(Debug, Default)]
pub struct PlanCache {
@@ -372,11 +323,12 @@ impl StagesGraph {
mod tests {
use super::{
JobStageResponse, JobStagesPopup, JobStagesResponse, StageTaskResponse,
- StageTaskStatus, StagesGraph, TaskPercentiles,
+ StageTaskStatus, StagesGraph,
};
+ use ballista_api_types::dto::Percentiles;
- fn make_percentiles() -> TaskPercentiles {
- TaskPercentiles {
+ fn make_percentiles() -> Percentiles {
+ Percentiles {
min: 0,
max: 100,
median: 50,
@@ -387,9 +339,9 @@ mod tests {
fn make_stage(id: &str) -> JobStageResponse {
JobStageResponse {
- id: id.to_string(),
- status: "Completed".to_string(),
- plan: String::new(),
+ stage_id: id.to_string(),
+ stage_status: "Completed".to_string(),
+ stage_plan: None,
input_rows: 0,
output_rows: 0,
elapsed_compute: Some("1ns".to_string()),
@@ -568,7 +520,7 @@ mod tests {
let mut popup = make_popup(3);
popup.table_state.select(Some(1));
let stage = popup.selected_stage().unwrap();
- assert_eq!(stage.id, "1");
+ assert_eq!(stage.stage_id, "1");
}
#[test]
@@ -611,19 +563,69 @@ mod tests {
assert_eq!(graph.scroll_position, 0);
}
+ /// Regression test for #2257. The scheduler serves `partition_id` as an
+ /// array of the global partitions a task owns, and the TUI used to declare
+ /// it as a scalar `u32`, so the whole stages response failed to parse. This
+ /// pins the shape against a payload shaped like the real one.
+ #[test]
+ fn deserializes_multi_partition_task_payload() {
+ let payload = r#"{
+ "stages": [{
+ "stage_id": "1",
+ "stage_status": "Successful",
+ "input_rows": 10,
+ "output_rows": 10,
+ "elapsed_compute": "1ms",
+ "tasks": [{
+ "id": 0,
+ "status": "Successful",
+ "partition_id": [0, 1, 2],
+ "scheduled_time": 0,
+ "launch_time": 0,
+ "start_exec_time": 0,
+ "end_exec_time": 5,
+ "exec_duration": 5,
+ "finish_time": 5,
+ "input_rows": 10,
+ "output_rows": 10
+ }]
+ }]
+ }"#;
+
+ let parsed: JobStagesResponse = serde_json::from_str(payload).unwrap();
+ let task = parsed.stages[0].tasks[0].as_ref().unwrap();
+ assert_eq!(task.partition_id, vec![0, 1, 2]);
+ }
+
+ /// The scheduler's failed-task status carries an `error` alongside
+ /// `reason`; the TUI's old copy only had `reason`.
+ #[test]
+ fn deserializes_failed_task_status_with_error() {
+ let payload = r#"{"Failed": {"reason": "ExecutionError", "error": "boom"}}"#;
+ let status: StageTaskStatus = serde_json::from_str(payload).unwrap();
+ match status {
+ StageTaskStatus::Failed { reason, error } => {
+ assert_eq!(reason, "ExecutionError");
+ assert_eq!(error, "boom");
+ }
+ other => panic!("expected Failed, got {other:?}"),
+ }
+ }
+
// --- Helpers for task-bearing stages ---
fn make_task(id: usize) -> StageTaskResponse {
StageTaskResponse {
id,
status: StageTaskStatus::Successful,
- partition_id: id as u32,
+ partition_id: vec![id as u32],
input_rows: 0,
output_rows: 0,
scheduled_time: 0,
launch_time: 0,
start_exec_time: 0,
end_exec_time: 0,
+ exec_duration: 0,
finish_time: 0,
}
}
diff --git a/ballista-cli/src/tui/mod.rs b/ballista-cli/src/tui/mod.rs
index 70e1df032..343cc7780 100644
--- a/ballista-cli/src/tui/mod.rs
+++ b/ballista-cli/src/tui/mod.rs
@@ -297,9 +297,9 @@ pub(crate) mod web {
.await
{
Ok(mut stages) => {
- stages
- .stages
- .sort_by_key(|s| s.id.parse::<u64>().unwrap_or(u64::MAX));
+ stages.stages.sort_by_key(|s| {
+ s.stage_id.parse::<u64>().unwrap_or(u64::MAX)
+ });
send_data(UiData::JobStagesData(id, stages), tx).await;
}
Err(e) => {
@@ -383,9 +383,9 @@ pub(crate) mod web {
WebKeyAsyncAction::LoadStagePlan(job_id, tab) => {
match http_client.get_job_stages(&job_id, &tab).await {
Ok(mut stages) => {
- stages
- .stages
- .sort_by_key(|s| s.id.parse::<u64>().unwrap_or(u64::MAX));
+ stages.stages.sort_by_key(|s| {
+ s.stage_id.parse::<u64>().unwrap_or(u64::MAX)
+ });
send_data(UiData::JobStagesPlanData(tab, stages), tx).await;
}
Err(e) => {
diff --git a/ballista-cli/src/tui/ui/main/jobs/job_stages_popup.rs b/ballista-cli/src/tui/ui/main/jobs/job_stages_popup.rs
index 383e5bf5c..1a164ad68 100644
--- a/ballista-cli/src/tui/ui/main/jobs/job_stages_popup.rs
+++ b/ballista-cli/src/tui/ui/main/jobs/job_stages_popup.rs
@@ -88,7 +88,7 @@ fn build_stage_row(i: usize, stage: &JobStageResponse, app: &App) -> Row<'static
app.theme.row_odd
};
- let status_style = match stage.status.as_str() {
+ let status_style = match stage.stage_status.as_str() {
"Running" => app.theme.status_running,
"Queued" => app.theme.status_queued,
"Successful" | "Completed" => app.theme.status_completed,
@@ -124,9 +124,9 @@ fn build_stage_row(i: usize, stage: &JobStageResponse, app: &App) -> Row<'static
);
Row::new(vec![
- Cell::from(Text::from(stage.id.clone()).right_aligned()),
+ Cell::from(Text::from(stage.stage_id.clone()).right_aligned()),
Cell::from(
- Text::from(stage.status.clone())
+ Text::from(stage.stage_status.clone())
.style(status_style)
.centered(),
),
diff --git a/ballista-cli/src/tui/ui/main/jobs/mod.rs b/ballista-cli/src/tui/ui/main/jobs/mod.rs
index 9db6c88e2..6880e5d2e 100644
--- a/ballista-cli/src/tui/ui/main/jobs/mod.rs
+++ b/ballista-cli/src/tui/ui/main/jobs/mod.rs
@@ -39,6 +39,8 @@ use crate::tui::{
ui::vertical_scrollbar::render_scrollbar,
};
+use crate::tui::app::INVALID_DATE;
+use crate::tui::domain::jobs::JobStatusExt;
use crate::tui::ui::components::clear_area::clear_area;
use crate::tui::ui::components::loading_indicator::shimmer_spans_with_style;
use crate::tui::ui::vertical_scrollbar;
@@ -118,7 +120,7 @@ pub async fn load_job_stages_popup(app: &App, job_id: &str) -> TuiResult<()> {
stages
.stages
- .sort_by_key(|s| s.id.parse::<u64>().unwrap_or(u64::MAX));
+ .sort_by_key(|s| s.stage_id.parse::<u64>().unwrap_or(u64::MAX));
app.send_event(Event::DataLoaded {
data: UiData::JobStagesData(job_id.to_owned(), stages),
@@ -144,7 +146,7 @@ pub async fn load_stage_plan(
stages
.stages
- .sort_by_key(|s| s.id.parse::<u64>().unwrap_or(u64::MAX));
+ .sort_by_key(|s| s.stage_id.parse::<u64>().unwrap_or(u64::MAX));
app.send_event(Event::DataLoaded {
data: UiData::JobStagesPlanData(tab.clone(), stages),
@@ -359,14 +361,17 @@ fn render_jobs_table(
}
fn render_job_start_time_cell<'a>(job: &'a Job, app: &App) -> Cell<'a> {
- let start_time = app.format_datetime(job.start_time);
+ let start_time = i64::try_from(job.start_time).map_or_else(
+ |_| INVALID_DATE.to_string(),
+ |time| app.format_datetime(time),
+ );
Cell::from(Text::from(start_time).centered())
}
fn render_job_duration_cell<'a>(job: &'a Job, app: &App) -> Cell<'a> {
let duration = if job.end_time > 0 {
- let duration = job.end_time - job.start_time;
- app.format_duration(duration as u64)
+ let duration = job.end_time.saturating_sub(job.start_time);
+ app.format_duration(duration)
} else {
"N/A".to_string()
};
diff --git a/ballista-cli/src/tui/ui/main/jobs/stage_plan_popup.rs b/ballista-cli/src/tui/ui/main/jobs/stage_plan_popup.rs
index ca9a75f0e..046e1e908 100644
--- a/ballista-cli/src/tui/ui/main/jobs/stage_plan_popup.rs
+++ b/ballista-cli/src/tui/ui/main/jobs/stage_plan_popup.rs
@@ -42,7 +42,10 @@ fn render_plans(f: &mut Frame, area: Rect, app: &App) {
return;
};
- let title = format!(" Plan for stage '{}' of job '{}' ", stage.id, popup.job_id);
+ let title = format!(
+ " Plan for stage '{}' of job '{}' ",
+ stage.stage_id, popup.job_id
+ );
let block = Block::default()
.title(title)
@@ -50,10 +53,12 @@ fn render_plans(f: &mut Frame, area: Rect, app: &App) {
.border_style(app.theme.popup_border)
.border_type(BorderType::Thick);
- let paragraph = Paragraph::new(&*stage.plan).block(block).scroll((
- popup.plan_vertical_scroll_position(),
- popup.plan_horizontal_scroll_position(),
- ));
+ let paragraph = Paragraph::new(stage.stage_plan.as_deref().unwrap_or_default())
+ .block(block)
+ .scroll((
+ popup.plan_vertical_scroll_position(),
+ popup.plan_horizontal_scroll_position(),
+ ));
f.render_widget(paragraph, area);
}
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 8c89a699b..cfe0a8329 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
@@ -77,7 +77,7 @@ pub(crate) fn render_stage_tasks_popup(f: &mut Frame, app: &App) {
Block::default()
.title(format!(
" Tasks for stage '{}' of job '{}' ",
- stage.id, popup.job_id
+ stage.stage_id, popup.job_id
))
.borders(Borders::ALL)
.border_style(app.theme.popup_border_alt)
@@ -106,8 +106,8 @@ fn build_stage_task_row(i: usize, task: &StageTaskResponse, app: &App) -> Row<'s
StageTaskStatus::Successful => {
Text::from("Successful").style(app.theme.status_completed)
}
- StageTaskStatus::Failed { reason } => {
- Text::from(reason.clone()).style(app.theme.status_failed)
+ StageTaskStatus::Failed { reason, error } => {
+ Text::from(format!("{reason}: {error}")).style(app.theme.status_failed)
}
};
@@ -116,7 +116,7 @@ fn build_stage_task_row(i: usize, task: &StageTaskResponse, app: &App) -> Row<'s
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()),
+ Cell::from(Text::from(format_partitions(&task.partition_id)).right_aligned()),
Cell::from(Text::from(format_datetime(task.scheduled_time, app)).centered()),
Cell::from(
Text::from(
@@ -149,3 +149,13 @@ fn build_stage_task_row(i: usize, task: &StageTaskResponse, app: &App) -> Row<'s
fn format_datetime(dt: u64, app: &App) -> String {
app.format_datetime(dt.try_into().unwrap_or(0))
}
+
+/// Render the global partitions a task owns. Single-partition tasks show just
+/// the id; multi-partition tasks show the whole list.
+fn format_partitions(partitions: &[u32]) -> String {
+ partitions
+ .iter()
+ .map(|p| p.to_string())
+ .collect::<Vec<_>>()
+ .join(", ")
+}
diff --git a/ballista/history/Cargo.toml b/ballista/history/Cargo.toml
index 009bf0172..092d927d2 100644
--- a/ballista/history/Cargo.toml
+++ b/ballista/history/Cargo.toml
@@ -27,7 +27,7 @@ edition = { workspace = true }
rust-version = { workspace = true }
[dependencies]
-ballista-api-types = { path = "../api-types", version = "54.0.0" }
+ballista-api-types = { workspace = true }
log = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { version = "1", features = ["raw_value"] }
diff --git a/ballista/scheduler/Cargo.toml b/ballista/scheduler/Cargo.toml
index 627e34498..c6038dbe5 100644
--- a/ballista/scheduler/Cargo.toml
+++ b/ballista/scheduler/Cargo.toml
@@ -49,7 +49,7 @@ substrait = ["dep:datafusion-substrait"]
arrow-flight = { workspace = true }
async-trait = { workspace = true }
axum = "0.8.9"
-ballista-api-types = { path = "../api-types", version = "54.0.0", optional = true }
+ballista-api-types = { workspace = true, optional = true }
ballista-core = { path = "../core", version = "54.0.0" }
ballista-history = { path = "../history", version = "54.0.0", optional = true }
clap = { workspace = true, optional = true }
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org