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 daebb3c92 feat(executor): log task execution duration in finished-task
message (#1999)
daebb3c92 is described below
commit daebb3c928aa784933f2221b45e231b2994639b9
Author: Andy Grove <[email protected]>
AuthorDate: Sat Jul 11 10:38:20 2026 -0600
feat(executor): log task execution duration in finished-task message (#1999)
Include the elapsed execution time in the "Finished task" log at both
task-dispatch sites (the executor_server push path and the execution_loop
pull path), so operators can read per-task durations directly from executor
logs without cross-referencing task-status reports.
---
ballista/executor/src/execution_loop.rs | 8 ++++++--
ballista/executor/src/executor_server.rs | 8 ++++++--
2 files changed, 12 insertions(+), 4 deletions(-)
diff --git a/ballista/executor/src/execution_loop.rs
b/ballista/executor/src/execution_loop.rs
index 77cd4c7c5..8a759f501 100644
--- a/ballista/executor/src/execution_loop.rs
+++ b/ballista/executor/src/execution_loop.rs
@@ -44,7 +44,7 @@ use std::any::Any;
use std::convert::TryInto;
use std::error::Error;
use std::sync::mpsc::{Receiver, Sender, TryRecvError};
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Instant, SystemTime, UNIX_EPOCH};
use std::{sync::Arc, time::Duration};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tonic::codegen::{Body, Bytes, StdError};
@@ -297,6 +297,7 @@ async fn run_received_task<T: 'static + AsLogicalPlan, U:
'static + AsExecutionP
partition_id: partition_id as usize,
};
+ let task_start = Instant::now();
let execution_result = match
AssertUnwindSafe(executor.execute_query_stage(
task_id as usize,
part.clone(),
@@ -314,7 +315,10 @@ async fn run_received_task<T: 'static + AsLogicalPlan, U:
'static + AsExecutionP
}
};
- info!("Finished task : [{task_identity}]");
+ info!(
+ "Finished task : [{task_identity}] in {:?}",
+ task_start.elapsed()
+ );
debug!("Task statistics: [{task_identity}] {execution_result:?}");
let plan_metrics = query_stage_exec.collect_plan_metrics();
diff --git a/ballista/executor/src/executor_server.rs
b/ballista/executor/src/executor_server.rs
index 8e7b02354..e65dbb0e0 100644
--- a/ballista/executor/src/executor_server.rs
+++ b/ballista/executor/src/executor_server.rs
@@ -27,7 +27,7 @@ use std::collections::HashMap;
use std::convert::TryInto;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use sysinfo::{MemoryRefreshKind, System};
use tokio::sync::mpsc;
@@ -407,6 +407,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> ExecutorServer<T,
info!("Execute task : [{task_identity}]");
+ let task_start = Instant::now();
let execution_result = self
.executor
.execute_query_stage(
@@ -416,7 +417,10 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan> ExecutorServer<T,
task_context,
)
.await;
- info!("Finished task : [{task_identity}]");
+ info!(
+ "Finished task : [{task_identity}] in {:?}",
+ task_start.elapsed()
+ );
debug!(
"Task [{task_identity}], execution statistics:
{execution_result:?}"
);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
