You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by mg...@apache.org on 2026/05/13 11:06:02 UTC
(datafusion-ballista) branch main updated: Minor improvements for #1675 (#1686)
This is an automated email from the ASF dual-hosted git repository.
martin-g 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 b01ee79bc Minor improvements for #1675 (#1686)
b01ee79bc is described below
commit b01ee79bc5e13303ffd973380d1a77b7934ab91d
Author: Martin Grigorov <ma...@users.noreply.github.com>
AuthorDate: Wed May 13 14:05:58 2026 +0300
Minor improvements for #1675 (#1686)
* Minor improvements for #1675
* Time::to_string() returns the unit
* Use .expect() with an error message if the system time is before Unix
epoch (highly unlikely)
* Make `elapsed_compute` optional
If the task is not finished yet then its value will be None.
---
ballista-cli/src/tui/domain/jobs/stages.rs | 4 +-
.../src/tui/ui/main/jobs/job_stages_popup.rs | 5 +-
ballista/core/src/utils.rs | 4 +-
ballista/scheduler/src/api/handlers.rs | 77 ++++++++++++----------
4 files changed, 52 insertions(+), 38 deletions(-)
diff --git a/ballista-cli/src/tui/domain/jobs/stages.rs b/ballista-cli/src/tui/domain/jobs/stages.rs
index b65f1462c..f4d6aaad2 100644
--- a/ballista-cli/src/tui/domain/jobs/stages.rs
+++ b/ballista-cli/src/tui/domain/jobs/stages.rs
@@ -33,7 +33,7 @@ pub struct JobStageResponse {
pub plan: String,
pub input_rows: usize,
pub output_rows: usize,
- pub elapsed_compute: String,
+ pub elapsed_compute: Option<String>,
pub task_duration_percentiles: TaskPercentiles,
pub task_input_percentiles: TaskPercentiles,
pub tasks: Vec<StageTaskResponse>,
@@ -207,7 +207,7 @@ mod tests {
plan: String::new(),
input_rows: 0,
output_rows: 0,
- elapsed_compute: "0".to_string(),
+ elapsed_compute: Some("1ns".to_string()),
task_duration_percentiles: make_percentiles(),
task_input_percentiles: make_percentiles(),
tasks: Vec::new(),
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 b2b696bc2..874b24e9f 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
@@ -125,7 +125,10 @@ fn build_stage_row(i: usize, stage: &JobStageResponse, app: &App) -> Row<'static
),
Cell::from(Text::from(app.format_count(stage.input_rows)).right_aligned()),
Cell::from(Text::from(app.format_count(stage.output_rows)).right_aligned()),
- Cell::from(Text::from(stage.elapsed_compute.clone()).right_aligned()),
+ Cell::from(
+ Text::from(stage.elapsed_compute.clone().unwrap_or("N/A".to_string()))
+ .right_aligned(),
+ ),
Cell::from(Text::from(input_percentiles).centered()),
Cell::from(Text::from(duration_percentiles).centered()),
])
diff --git a/ballista/core/src/utils.rs b/ballista/core/src/utils.rs
index 68e98ea3c..842888468 100644
--- a/ballista/core/src/utils.rs
+++ b/ballista/core/src/utils.rs
@@ -349,11 +349,11 @@ pub fn get_time_before(interval_seconds: u64) -> u64 {
.as_secs()
}
-/// current time since UNIX EPOCH
+/// Current time since UNIX EPOCH. In milliseconds.
pub fn get_current_time() -> u128 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
- .unwrap()
+ .expect("system clock is before UNIX epoch")
.as_millis()
}
diff --git a/ballista/scheduler/src/api/handlers.rs b/ballista/scheduler/src/api/handlers.rs
index 925fbf7ea..fc8871bec 100644
--- a/ballista/scheduler/src/api/handlers.rs
+++ b/ballista/scheduler/src/api/handlers.rs
@@ -197,7 +197,7 @@ pub struct QueryStageSummary {
pub stage_status: String,
pub input_rows: usize,
pub output_rows: usize,
- pub elapsed_compute: String,
+ pub elapsed_compute: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub stage_plan: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -522,7 +522,7 @@ pub async fn get_query_stages<
stage_status: stage.variant_name().to_string(),
input_rows: 0,
output_rows: 0,
- elapsed_compute: "".to_string(),
+ elapsed_compute: None,
tasks: vec![],
task_duration_percentiles: None,
task_input_percentiles: None,
@@ -546,8 +546,7 @@ pub async fn get_query_stages<
.map(|m| get_combined_count(m.as_slice(), "output_rows"))
.unwrap_or(0);
summary.elapsed_compute = get_running_stage_time(&running_stage
- .task_infos, get_current_time())
- ;
+ .task_infos, get_current_time());
summary.tasks = running_stage
.task_infos
.iter()
@@ -734,7 +733,10 @@ fn format_job_status(status: &Option<Status>, elapsed_ms: u64) -> (String, Strin
}
}
-fn get_running_stage_time(task_infos: &[Option<TaskInfo>], current_time: u128) -> String {
+fn get_running_stage_time(
+ task_infos: &[Option<TaskInfo>],
+ current_time: u128,
+) -> Option<String> {
let min_start = task_infos
.iter()
.flat_map(|t| t.as_ref().map(|t| t.start_exec_time))
@@ -742,16 +744,16 @@ fn get_running_stage_time(task_infos: &[Option<TaskInfo>], current_time: u128) -
.min();
match (min_start, current_time) {
- (Some(start), end) if end > start => {
- let t = Time::new();
- t.add_duration(Duration::from_millis((end - start) as u64));
- t.to_string()
+ (Some(start), end) if end >= start => {
+ let time = Time::new();
+ time.add_duration(Duration::from_millis((end - start) as u64));
+ Some(time.to_string())
}
- _ => "0".to_string(),
+ _ => None,
}
}
-fn get_finished_stage_time(task_infos: &[TaskInfo]) -> String {
+fn get_finished_stage_time(task_infos: &[TaskInfo]) -> Option<String> {
let min_start = task_infos
.iter()
.map(|t| t.start_exec_time)
@@ -765,12 +767,12 @@ fn get_finished_stage_time(task_infos: &[TaskInfo]) -> String {
.max();
match (min_start, max_end) {
- (Some(start), Some(end)) if end > start => {
- let t = Time::new();
- t.add_duration(Duration::from_millis((end - start) as u64));
- t.to_string()
+ (Some(start), Some(end)) if end >= start => {
+ let time = Time::new();
+ time.add_duration(Duration::from_millis((end - start) as u64));
+ Some(time.to_string())
}
- _ => "0".to_string(),
+ _ => None,
}
}
@@ -937,21 +939,24 @@ mod tests {
// --- get_finished_stage_time ---
#[test]
- fn test_finished_empty_slice_returns_zero() {
- assert_eq!(get_finished_stage_time(&[]), "0");
+ fn test_finished_empty_slice_returns_none() {
+ assert_eq!(get_finished_stage_time(&[]), None);
}
#[test]
- fn test_finished_all_zero_timestamps_returns_zero() {
+ fn test_finished_all_zero_timestamps_returns_none() {
let tasks = vec![make_task_info(0, 0), make_task_info(0, 0)];
- assert_eq!(get_finished_stage_time(&tasks), "0");
+ assert_eq!(get_finished_stage_time(&tasks), None);
}
#[test]
fn test_finished_single_task_elapsed() {
// 600 - 100 = 500 ms → "500.00ms"
let tasks = vec![make_task_info(100, 600)];
- assert_eq!(get_finished_stage_time(&tasks), "500.00ms");
+ assert_eq!(
+ get_finished_stage_time(&tasks),
+ Some("500.00ms".to_string())
+ );
}
#[test]
@@ -962,41 +967,47 @@ mod tests {
make_task_info(200, 900),
make_task_info(300, 700),
];
- assert_eq!(get_finished_stage_time(&tasks), "800.00ms");
+ assert_eq!(
+ get_finished_stage_time(&tasks),
+ Some("800.00ms".to_string())
+ );
}
#[test]
- fn test_finished_end_before_start_returns_zero() {
+ fn test_finished_end_before_start_returns_none() {
let tasks = vec![make_task_info(900, 100)];
- assert_eq!(get_finished_stage_time(&tasks), "0");
+ assert_eq!(get_finished_stage_time(&tasks), None);
}
// --- get_running_stage_time ---
#[test]
- fn test_running_empty_slice_returns_zero() {
- assert_eq!(get_running_stage_time(&[], 1000), "0");
+ fn test_running_empty_slice_returns_none() {
+ assert_eq!(get_running_stage_time(&[], 1000), None);
}
#[test]
- fn test_running_all_none_returns_zero() {
+ fn test_running_all_none_returns_none() {
let tasks: Vec<Option<TaskInfo>> = vec![None, None];
- assert_eq!(get_running_stage_time(&tasks, 1000), "0");
+ assert_eq!(get_running_stage_time(&tasks, 1000), None);
}
#[test]
- fn test_running_future_start_returns_zero() {
+ fn test_running_future_start_returns_none() {
// start_exec_time beyond current time → elapsed clamped to 0
let tasks = vec![Some(make_task_info(u128::MAX, 0))];
- assert_eq!(get_running_stage_time(&tasks, 1000), "0");
+ assert_eq!(get_running_stage_time(&tasks, 1000), None);
}
#[test]
- fn test_running_past_start_returns_nonzero() {
+ fn test_running_past_start_returns_some() {
let now = 4_000;
let start = 1_000;
let tasks = vec![Some(make_task_info(start, 0))];
- assert_eq!(get_running_stage_time(&tasks, now), "3.00s");
+ assert_eq!(
+ get_running_stage_time(&tasks, now),
+ Some("3.00s".to_string())
+ );
}
#[test]
@@ -1011,7 +1022,7 @@ mod tests {
None,
];
let result = get_running_stage_time(&tasks, now);
- assert_eq!(result, "2.00s");
+ assert_eq!(result, Some("2.00s".to_string()));
}
#[test]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org