Add “skipped stages” table; only display non-empty tables.

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

@@ -51,17 +51,14 @@ private[ui] class JobPage(parent: JobsTab) extends WebUIPage("job") {
           new StageInfo(stageId, 0, "Unknown", 0, Seq.empty, "Unknown"))
       }
 
-      val pendingStages = mutable.Buffer[StageInfo]()
       val activeStages = mutable.Buffer[StageInfo]()
       val completedStages = mutable.Buffer[StageInfo]()
+      // If the job is completed, then any pending stages are displayed as "skipped":
+      val pendingOrSkippedStages = mutable.Buffer[StageInfo]()
       val failedStages = mutable.Buffer[StageInfo]()
       for (stage <- stages) {
         if (stage.submissionTime.isEmpty) {
-          if (!isComplete) {
-            pendingStages += stage
-          } else {
-            // Do nothing so that we don't display pending stages for completed jobs
-          }
+          pendingOrSkippedStages += stage
         } else if (stage.completionTime.isDefined) {
           if (stage.failureReason.isDefined) {
             failedStages += stage
@@ -77,8 +74,8 @@ private[ui] class JobPage(parent: JobsTab) extends WebUIPage("job") {
         new StageTableBase(activeStages.sortBy(_.submissionTime).reverse,
           parent.basePath, parent.listener, isFairScheduler = parent.isFairScheduler,
           killEnabled = parent.killEnabled)
-      val pendingStagesTable =
-        new StageTableBase(pendingStages.sortBy(_.stageId).reverse,
+      val pendingOrSkippedStagesTable =
+        new StageTableBase(pendingOrSkippedStages.sortBy(_.stageId).reverse,
           parent.basePath, parent.listener, isFairScheduler = parent.isFairScheduler,
           killEnabled = false)
       val completedStagesTable =
@@ -89,6 +86,12 @@ private[ui] class JobPage(parent: JobsTab) extends WebUIPage("job") {
         new FailedStageTable(failedStages.sortBy(_.submissionTime).reverse, parent.basePath,
           parent.listener, isFairScheduler = parent.isFairScheduler)
 
+      val shouldShowActiveStages = activeStages.nonEmpty
+      val shouldShowPendingStages = !isComplete && pendingOrSkippedStages.nonEmpty
+      val shouldShowCompletedStages = completedStages.nonEmpty
+      val shouldShowSkippedStages = isComplete && pendingOrSkippedStages.nonEmpty
+      val shouldShowFailedStages = failedStages.nonEmpty
+
       val summary: NodeSeq =
         <div>
           <ul class="unstyled">
@@ -102,36 +105,73 @@ private[ui] class JobPage(parent: JobsTab) extends WebUIPage("job") {
                   <strong>Job Group:</strong>
                   {jobData.jobGroup.get}
                 </li>
-              } else Seq.empty
+              }
+            }
+            {
+              if (shouldShowActiveStages) {
+                <li>
+                  <a href="#active"><strong>Active Stages:</strong></a>
+                  {activeStages.size}
+                </li>
+              }
+            }
+            {
+              if (shouldShowPendingStages) {
+                <li>
+                  <a href="#pending">
+                    <strong>Pending Stages:</strong>
+                  </a>{pendingOrSkippedStages.size}
+                </li>
+              }
+            }
+            {
+              if (shouldShowCompletedStages) {
+                <li>
+                  <a href="#completed"><strong>Completed Stages:</strong></a>
+                  {completedStages.size}
+                </li>
+              }
+            }
+            {
+              if (shouldShowSkippedStages) {
+              <li>
+                <a href="#skipped"><strong>Skipped Stages:</strong></a>
+                {pendingOrSkippedStages.size}
+              </li>
+            }
+            }
+            {
+              if (shouldShowFailedStages) {
+                <li>
+                  <a href="#failed"><strong>Failed Stages:</strong></a>
+                  {failedStages.size}
+                </li>
+              }
             }
-            <li>
-              <a href="#active"><strong>Active Stages:</strong></a>
-              {activeStages.size}
-            </li>
-            <li>
-              <a href="#pending"><strong>Pending Stages:</strong></a>
-              {pendingStages.size}
-            </li>
-            <li>
-              <a href="#completed"><strong>Completed Stages:</strong></a>
-              {completedStages.size}
-            </li>
-            <li>
-              <a href="#failed"><strong>Failed Stages:</strong></a>
-              {failedStages.size}
-            </li>
           </ul>
         </div>
 
-      val content = summary ++
-        <h4 id="active">Active Stages ({activeStages.size})</h4> ++
-        activeStagesTable.toNodeSeq ++
-        <h4 id="pending">Pending Stages ({pendingStages.size})</h4> ++
-        pendingStagesTable.toNodeSeq ++
-        <h4 id="completed">Completed Stages ({completedStages.size})</h4> ++
-        completedStagesTable.toNodeSeq ++
-        <h4 id ="failed">Failed Stages ({failedStages.size})</h4> ++
-        failedStagesTable.toNodeSeq
+      var content = summary
+      if (shouldShowActiveStages) {
+        content ++= <h4 id="active">Active Stages ({activeStages.size})</h4> ++
+          activeStagesTable.toNodeSeq
+      }
+      if (shouldShowPendingStages) {
+        content ++= <h4 id="pending">Pending Stages ({pendingOrSkippedStages.size})</h4> ++
+          pendingOrSkippedStagesTable.toNodeSeq
+      }
+      if (shouldShowCompletedStages) {
+        content ++= <h4 id="completed">Completed Stages ({completedStages.size})</h4> ++
+          completedStagesTable.toNodeSeq
+      }
+      if (shouldShowSkippedStages) {
+        content ++= <h4 id="skipped">Skipped Stages ({pendingOrSkippedStages.size})</h4> ++
+          pendingOrSkippedStagesTable.toNodeSeq
+      }
+      if (shouldShowFailedStages) {
+        content ++= <h4 id ="failed">Failed Stages ({failedStages.size})</h4> ++
+          failedStagesTable.toNodeSeq
+      }
       UIUtils.headerSparkPage(s"Details for Job $jobId", content, parent)
     }
   }

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

@@ -58,6 +58,7 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
   // Stages:
   val activeStages = new HashMap[StageId, StageInfo]
   val completedStages = ListBuffer[StageInfo]()
+  val skippedStages = ListBuffer[StageInfo]()
   val failedStages = ListBuffer[StageInfo]()
   val stageIdToData = new HashMap[(StageId, StageAttemptId), StageUIData]
   val stageIdToInfo = new HashMap[StageId, StageInfo]
@@ -106,6 +107,7 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
       "completedJobs" -> completedJobs.size,
       "failedJobs" -> failedJobs.size,
       "completedStages" -> completedStages.size,
+      "skippedStages" -> skippedStages.size,
       "failedStages" -> failedStages.size
     )
   }
@@ -199,13 +201,10 @@ class JobProgressListener(conf: SparkConf) extends SparkListener with Logging {
       stageIdToActiveJobIds.get(stageId).foreach { jobsUsingStage =>
         jobsUsingStage.remove(jobEnd.jobId)
         stageIdToInfo.get(stageId).foreach { stageInfo =>
-          // If this is a pending stage and no other job depends on it, then it won't be run.
-          // To prevent memory leaks, remove this data since it won't be cleaned up as stages
-          // finish / fail:
-          val isPendingStage = stageInfo.submissionTime.isEmpty && stageInfo.completionTime.isEmpty
-          if (isPendingStage && jobsUsingStage.isEmpty) {
-            stageIdToInfo.remove(stageId)
-            stageIdToData.remove((stageId, stageInfo.attemptId))
+          if (stageInfo.submissionTime.isEmpty) {
+            // if this stage is pending, it won't complete, so mark it as "skipped":
+            skippedStages += stageInfo
+            trimStagesIfNecessary(skippedStages)
           }
         }
       }

core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala

@@ -210,6 +210,8 @@ class UISeleniumSuite extends FunSuite with WebBrowser with Matchers {
       rdd.countAsync()
       eventually(timeout(10 seconds), interval(50 milliseconds)) {
         go to (sc.ui.get.appUIAddress.stripSuffix("/") + "/jobs/job/?id=0")
+        find(id("active")).get.text should be ("Active Stages (1)")
+        find(id("pending")).get.text should be ("Pending Stages (2)")
         // Essentially, we want to check that none of the stage rows show
         // "No data available for this stage". Checking for the absence of that string is brittle
         // because someone could change the error message and cause this test to pass by accident.
@@ -222,7 +224,7 @@ class UISeleniumSuite extends FunSuite with WebBrowser with Matchers {
     }
   }
 
-  test("stages that aren't run do not show up in 'pending stages' after a job finishes") {
+  test("stages that aren't run appear as 'skipped stages' after a job finishes") {
     withSpark(newSparkContext()) { sc =>
       // Create an RDD that involves multiple stages:
       val rdd =
@@ -233,15 +235,23 @@ class UISeleniumSuite extends FunSuite with WebBrowser with Matchers {
       rdd.count()
       eventually(timeout(10 seconds), interval(50 milliseconds)) {
         go to (sc.ui.get.appUIAddress.stripSuffix("/") + "/jobs/job/?id=1")
-        find(id("pending")).get.text should be ("Pending Stages (0)")
-        find(id("active")).get.text should be ("Active Stages (0)")
+        find(id("pending")) should be (None)
+        find(id("active")) should be (None)
+        find(id("failed")) should be (None)
         find(id("completed")).get.text should be ("Completed Stages (1)")
-        find(id("failed")).get.text should be ("Failed Stages (0)")
+        find(id("skipped")).get.text should be ("Skipped Stages (2)")
+        // Essentially, we want to check that none of the stage rows show
+        // "No data available for this stage". Checking for the absence of that string is brittle
+        // because someone could change the error message and cause this test to pass by accident.
+        // Instead, it's safer to check that each row contains a link to a stage details page.
+        findAll(cssSelector("tbody tr")).foreach { row =>
+          val link = row.underlying.findElement(By.xpath(".//a"))
+          link.getAttribute("href") should include("stage")
+        }
       }
     }
   }
 
-
   test("jobs with stages that are skipped should show correct link descriptions on all jobs page") {
     withSpark(newSparkContext()) { sc =>
       // Create an RDD that involves multiple stages: