[SPARK-24851] : Avoiding extra store lookup and adding href links

Author: pgandhi

core/src/main/scala/org/apache/spark/status/AppStatusStore.scala

@@ -94,12 +94,6 @@ private[spark] class AppStatusStore(
       }.toSeq
   }
 
-  def getJobIdsAssociatedWithStage(stageId: Int, stageAttemptId: Int): Seq[Int] = {
-    val stageKey = Array(stageId, stageAttemptId)
-    val jobIds = store.read(classOf[StageDataWrapper], stageKey).jobIds.toSeq
-    jobIds
-  }
-
   def lastStageAttempt(stageId: Int): v1.StageData = {
     val it = store.view(classOf[StageDataWrapper])
       .index("stageId")
@@ -118,10 +112,13 @@ private[spark] class AppStatusStore(
     }
   }
 
-  def stageAttempt(stageId: Int, stageAttemptId: Int, details: Boolean = false): v1.StageData = {
+  def stageAttempt(stageId: Int, stageAttemptId: Int, details: Boolean = false): StageDataWrapper = {
     val stageKey = Array(stageId, stageAttemptId)
-    val stage = store.read(classOf[StageDataWrapper], stageKey).info
-    if (details) stageWithDetails(stage) else stage
+    val stageDataWrapper: StageDataWrapper = store.read(classOf[StageDataWrapper], stageKey)
+    val stage = if (details) stageWithDetails(stageDataWrapper.info) else stageDataWrapper.info
+    val jobIds = stageDataWrapper.jobIds
+    val returnStageDataWrapper: StageDataWrapper = new StageDataWrapper(stage, jobIds, null)
+    returnStageDataWrapper
   }
 
   def taskCount(stageId: Int, stageAttemptId: Int): Long = {

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

@@ -56,7 +56,7 @@ private[v1] class StagesResource extends BaseAppResource {
       @PathParam("stageAttemptId") stageAttemptId: Int,
       @QueryParam("details") @DefaultValue("true") details: Boolean): StageData = withUI { ui =>
     try {
-      ui.store.stageAttempt(stageId, stageAttemptId, details = details)
+      ui.store.stageAttempt(stageId, stageAttemptId, details = details).info
     } catch {
       case _: NoSuchElementException =>
         // Change the message depending on whether there are any attempts for the requested stage.

core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala

@@ -105,8 +105,9 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
     val stageAttemptId = parameterAttempt.toInt
 
     val stageHeader = s"Details for Stage $stageId (Attempt $stageAttemptId)"
+    val stageDataWrapper = parent.store.stageAttempt(stageId, stageAttemptId, details = false)
     val stageData = parent.store
-      .asOption(parent.store.stageAttempt(stageId, stageAttemptId, details = false))
+      .asOption(stageDataWrapper.info)
       .getOrElse {
         val content =
           <div id="no-info">
@@ -115,7 +116,7 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
         return UIUtils.headerSparkPage(request, stageHeader, content, parent)
       }
 
-    val stageJobIds = parent.store.getJobIdsAssociatedWithStage(stageId, stageAttemptId)
+    val stageJobIds = stageDataWrapper.jobIds.toSeq
 
     val localitySummary = store.localitySummary(stageData.stageId, stageData.attemptId)
 
@@ -187,7 +188,10 @@ private[ui] class StagePage(parent: StagesTab, store: AppStatusStore) extends We
           {if (!stageJobIds.isEmpty) {
             <li>
               <strong>Associated Job Ids: </strong>
-              {stageJobIds}
+              {for(jobId <- stageJobIds) yield {val detailUrl = "%s/jobs/job/?id=%s".format(
+                UIUtils.prependBaseUri(request, parent.basePath), jobId)
+              <a href={s"${detailUrl}"}>{s"${jobId}"} &nbsp;&nbsp;</a>
+              }}
             </li>
           }}
         </ul>
@@ -1055,7 +1059,7 @@ private[ui] object ApiHelper {
   }
 
   def lastStageNameAndDescription(store: AppStatusStore, job: JobData): (String, String) = {
-    val stage = store.asOption(store.stageAttempt(job.stageIds.max, 0))
+    val stage = store.asOption(store.stageAttempt(job.stageIds.max, 0).info)
     (stage.map(_.name).getOrElse(""), stage.flatMap(_.description).getOrElse(job.name))
   }