[SPARK-58552][SQL][UI] Add total task time column to the SQL / DataFrame tab

sql/core/src/main/resources/org/apache/spark/sql/execution/ui/static/allexecutionspage.js

@@ -17,7 +17,7 @@
 
 /* global $, uiRoot, appBasePath, createSqlApiBase, getSqlTableColumns,
    withResolvedAppId, statusBadge, jobIdLinks, formatDurationSql,
-   descriptionHtml */
+   formatTotalTaskTime, descriptionHtml */
 
 $(document).ready(function () {
   // Read the cluster-level grouping toggle rendered into the page by Scala
@@ -132,7 +132,8 @@ $(document).ready(function () {
         var html = '<table id="' + childId +
           '" class="table table-sm table-bordered mb-0 sub-exec-table">';
         html += '<thead><tr><th>ID</th><th>Status</th><th>Description</th>' +
-          '<th>Duration</th><th>Succeeded Jobs</th></tr></thead><tbody>';
+          '<th>Duration</th><th>Total Task Time</th>' +
+          '<th>Succeeded Jobs</th></tr></thead><tbody>';
         subs.forEach(function (child) {
           html += '<tr><td><a href="' + basePath + '/SQL/execution/?id=' +
             child.id + '">' + child.id + '</a></td>';
@@ -141,6 +142,7 @@ $(document).ready(function () {
             id: child.id, description: child.description || ""
           }) + '</td>';
           html += '<td>' + formatDurationSql(child.duration) + '</td>';
+          html += '<td>' + formatTotalTaskTime(child.totalTaskTime) + '</td>';
           html += '<td>' + jobIdLinks(child.jobIds || []) + '</td></tr>';
         });
         html += '</tbody></table>';

sql/core/src/main/resources/org/apache/spark/sql/execution/ui/static/executionpage.js

@@ -43,6 +43,7 @@ $(document).ready(function () {
         description: data.description || "",
         submissionTime: data.submissionTime,
         duration: data.duration,
+        totalTaskTime: data.totalTaskTime,
         jobIds: data.successJobIds || [],
         errorMessage: data.errorMessage || ""
       };

sql/core/src/main/resources/org/apache/spark/sql/execution/ui/static/sql-table-utils.js

@@ -30,6 +30,14 @@ function formatDurationSql(milliseconds) {
   return hours.toFixed(1) + " h";
 }
 
+// Format a total task time value. A negative or absent value means "unknown"
+// (e.g. the execution has no stages to aggregate), which is shown as "N/A"
+// rather than a misleading "0 ms".
+function formatTotalTaskTime(value) {
+  if (value === null || value === undefined || value < 0) return "N/A";
+  return formatDurationSql(value);
+}
+
 function formatDateSql(dateStr) {
   if (!dateStr) return "";
   try {
@@ -232,6 +240,14 @@ function getSqlTableColumns(opts) {
     }
   };
 
+  var totalTaskTimeColumn = {
+    data: "totalTaskTime", name: "totalTaskTime", title: "Total Task Time",
+    render: function (data, type) {
+      if (type !== "display") return data;
+      return formatTotalTaskTime(data);
+    }
+  };
+
   var jobsColumn = {
     data: "jobIds", name: "jobIds", title: "Succeeded Jobs",
     orderable: false,
@@ -258,5 +274,6 @@ function getSqlTableColumns(opts) {
   };
 
   return [idColumn, queryIdColumn, statusColumn, descriptionColumn,
-    submissionColumn, durationColumn, jobsColumn, errorColumn];
+    submissionColumn, durationColumn, totalTaskTimeColumn, jobsColumn,
+    errorColumn];
 }

sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/SqlResource.scala

@@ -28,6 +28,7 @@ import jakarta.ws.rs.core.{Context, MediaType, UriInfo}
 import org.apache.spark.JobExecutionStatus
 import org.apache.spark.internal.config.UI.UI_SQL_GROUP_SUB_EXECUTION_ENABLED
 import org.apache.spark.sql.execution.ui.{SparkPlanGraph, SparkPlanGraphCluster, SparkPlanGraphNode, SQLAppStatusStore, SQLExecutionUIData}
+import org.apache.spark.status.AppStatusStore
 import org.apache.spark.status.api.v1.{BaseAppResource, NotFoundException}
 import org.apache.spark.ui.UIUtils
 
@@ -51,7 +52,7 @@ private[v1] class SqlResource extends BaseAppResource {
       }
       execs.map { exec =>
         val graph = sqlStore.planGraph(exec.executionId)
-        prepareExecutionData(exec, graph, details, planDescription)
+        prepareExecutionData(exec, graph, details, planDescription, ui.store)
       }
     }
   }
@@ -67,7 +68,10 @@ private[v1] class SqlResource extends BaseAppResource {
       val sqlStore = new SQLAppStatusStore(ui.store.store)
       sqlStore
         .execution(execId)
-        .map(prepareExecutionData(_, sqlStore.planGraph(execId), details, planDescription))
+        .map { exec =>
+          prepareExecutionData(exec, sqlStore.planGraph(execId), details, planDescription,
+            ui.store)
+        }
         .getOrElse(throw new NotFoundException("unknown query execution id: " + execId))
     }
   }
@@ -146,18 +150,31 @@ private[v1] class SqlResource extends BaseAppResource {
       val start = Option(uriParams.getFirst("start")).map(_.toInt).getOrElse(0)
       val length = Option(uriParams.getFirst("length")).map(_.toInt).getOrElse(20)
 
-      val sortedRoots = sortExecs(rootRows, sortCol, sortDir)
+      // Precompute the total task time of every root row only when the list is
+      // sorted by it, so the sort and the page rows reuse the same values
+      // instead of recomputing per stage attempt. When sorting by another
+      // column, `execToRow` computes it only for the rows on the current page.
+      val totalTaskTimeMap: Map[Long, Long] =
+        if (sortCol == "totalTaskTime") {
+          rootRows.iterator.map(e => e.executionId -> totalTaskTime(e, ui.store)).toMap
+        } else {
+          Map.empty
+        }
+
+      val sortedRoots = sortExecs(rootRows, sortCol, sortDir, totalTaskTimeMap)
       val page = if (length > 0) sortedRoots.slice(start, start + length) else sortedRoots
 
       // Convert to Java-compatible row data; embed sub-executions when grouping.
       // Always emit a `subExecutions` field (possibly empty) in grouped mode so
       // JSON consumers see a consistent schema; flat mode never includes it.
       val aaData = page.map { exec =>
-        val row = execToRow(exec)
+        val row = execToRow(exec, totalTaskTimeMap, ui.store)
         if (groupSubExec) {
           val subs = subsByRoot.getOrElse(exec.executionId, Seq.empty)
           // Sort subs by id ascending so they appear in chronological order
-          row.put("subExecutions", sortExecs(subs, "id", "asc").map(execToRow).asJava)
+          row.put("subExecutions",
+            sortExecs(subs, "id", "asc", totalTaskTimeMap)
+              .map(execToRow(_, totalTaskTimeMap, ui.store)).asJava)
         }
         row
       }
@@ -191,7 +208,8 @@ private[v1] class SqlResource extends BaseAppResource {
   private def sortExecs(
       execs: Seq[SQLExecutionUIData],
       sortCol: String,
-      sortDir: String): Seq[SQLExecutionUIData] = {
+      sortDir: String,
+      totalTaskTimeMap: Map[Long, Long]): Seq[SQLExecutionUIData] = {
     val sorted = sortCol match {
       case "id" => execs.sortBy(_.executionId)
       case "status" => execs.sortBy(_.executionStatus)
@@ -200,12 +218,36 @@ private[v1] class SqlResource extends BaseAppResource {
       case "duration" =>
         execs.sortBy(e =>
           e.completionTime.getOrElse(new Date()).getTime - e.submissionTime)
+      case "totalTaskTime" =>
+        execs.sortBy(e => totalTaskTimeMap.getOrElse(e.executionId, -1L))
       case _ => execs.sortBy(_.executionId)
     }
     if (sortDir == "asc") sorted else sorted.reverse
   }
 
-  private def execToRow(exec: SQLExecutionUIData): java.util.LinkedHashMap[String, Object] = {
+  /**
+   * Total task time of an execution, in milliseconds, aggregated across all
+   * stages of the execution. Sums `executorRunTime` (the cumulative time
+   * executors spent running tasks, which is the "Total Time Across All Tasks"
+   * stage-level metric) of every attempt of every stage: each attempt
+   * genuinely consumed task time, including failed attempts that were
+   * retried. Returns -1 when the execution has no stages to aggregate, so
+   * callers can distinguish "no task time information" from a genuine zero.
+   */
+  private def totalTaskTime(exec: SQLExecutionUIData, store: AppStatusStore): Long = {
+    if (exec.stages.isEmpty) {
+      -1L
+    } else {
+      exec.stages.iterator.flatMap { stageId =>
+        store.stageData(stageId).map(_.executorRunTime)
+      }.sum
+    }
+  }
+
+  private def execToRow(
+      exec: SQLExecutionUIData,
+      totalTaskTimeMap: Map[Long, Long],
+      store: AppStatusStore): java.util.LinkedHashMap[String, Object] = {
     val duration = exec.completionTime.getOrElse(new Date()).getTime - exec.submissionTime
     val jobIds = exec.jobs.collect {
       case (id, JobExecutionStatus.SUCCEEDED) => id
@@ -216,6 +258,8 @@ private[v1] class SqlResource extends BaseAppResource {
     row.put("description", exec.description)
     row.put("submissionTime", new Date(exec.submissionTime))
     row.put("duration", java.lang.Long.valueOf(duration))
+    row.put("totalTaskTime", java.lang.Long.valueOf(
+      totalTaskTimeMap.getOrElse(exec.executionId, totalTaskTime(exec, store))))
     row.put("jobIds", jobIds)
     row.put("queryId", if (exec.queryId != null) exec.queryId.toString else null)
     row.put("errorMessage", exec.errorMessage.orNull)
@@ -227,7 +271,8 @@ private[v1] class SqlResource extends BaseAppResource {
     exec: SQLExecutionUIData,
     graph: SparkPlanGraph,
     details: Boolean,
-    planDescription: Boolean): ExecutionData = {
+    planDescription: Boolean,
+    store: AppStatusStore): ExecutionData = {
 
     var running = Seq[Int]()
     var completed = Seq[Int]()
@@ -267,7 +312,8 @@ private[v1] class SqlResource extends BaseAppResource {
       if (exec.queryId != null) exec.queryId.toString else null,
       exec.errorMessage.orNull,
       exec.rootExecutionId,
-      exec.modifiedConfigs)
+      exec.modifiedConfigs,
+      totalTaskTime(exec, store))
   }
 
   private def printableMetrics(allNodes: collection.Seq[SparkPlanGraphNode],

sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/api.scala

@@ -44,4 +44,5 @@ class ExecutionData private[spark] (
     val queryId: String = null,
     val errorMessage: String = null,
     val rootExecutionId: Long = -1,
-    val modifiedConfigs: Map[String, String] = Map.empty)
+    val modifiedConfigs: Map[String, String] = Map.empty,
+    val totalTaskTime: Long = -1L)