[SPARK-26260][Core]For disk store tasks summary table should show only successful tasks summary

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

@@ -136,12 +136,6 @@ private[spark] class AppStatusStore(
     store.read(classOf[StageDataWrapper], Array(stageId, stageAttemptId)).locality
   }
 
-  // SPARK-26119: we only want to consider successful tasks when calculating the metrics summary,
-  // but currently this is very expensive when using a disk store. So we only trigger the slower
-  // code path when we know we have all data in memory. The following method checks whether all
-  // the data will be in memory.
-  private def isInMemoryStore: Boolean = store.isInstanceOf[InMemoryStore] || listener.isDefined
-
   /**
    * Calculates a summary of the task metrics for the given stage attempt, returning the
    * requested quantiles for the recorded metrics.
@@ -162,21 +156,11 @@ private[spark] class AppStatusStore(
     // cheaper for disk stores (avoids deserialization).
     val count = {
       Utils.tryWithResource(
-        if (isInMemoryStore) {
-          // For Live UI, we should count the tasks with status "SUCCESS" only.
-          store.view(classOf[TaskDataWrapper])
-            .parent(stageKey)
-            .index(TaskIndexNames.STATUS)
-            .first("SUCCESS")
-            .last("SUCCESS")
-            .closeableIterator()
-        } else {
-          store.view(classOf[TaskDataWrapper])
-            .parent(stageKey)
-            .index(TaskIndexNames.EXEC_RUN_TIME)
-            .first(0L)
-            .closeableIterator()
-        }
+        store.view(classOf[TaskDataWrapper])
+          .parent(stageKey)
+          .index(SuccessTaskIndexNames.EXEC_RUN_TIME)
+          .first(0L)
+          .closeableIterator()
       ) { it =>
         var _count = 0L
         while (it.hasNext()) {
@@ -245,100 +229,82 @@ private[spark] class AppStatusStore(
     // stabilize once the stage finishes. It's also slow, especially with disk stores.
     val indices = quantiles.map { q => math.min((q * count).toLong, count - 1) }
 
-    // TODO: Summary metrics needs to display all the successful tasks' metrics (SPARK-26119).
-    // For InMemory case, it is efficient to find using the following code. But for diskStore case
-    // we need an efficient solution to avoid deserialization time overhead. For that, we need to
-    // rework on the way indexing works, so that we can index by specific metrics for successful
-    // and failed tasks differently (would be tricky). Also would require changing the disk store
-    // version (to invalidate old stores).
     def scanTasks(index: String)(fn: TaskDataWrapper => Long): IndexedSeq[Double] = {
-      if (isInMemoryStore) {
-        val quantileTasks = store.view(classOf[TaskDataWrapper])
+      Utils.tryWithResource(
+        store.view(classOf[TaskDataWrapper])
           .parent(stageKey)
           .index(index)
           .first(0L)
-          .asScala
-          .filter { _.status == "SUCCESS"} // Filter "SUCCESS" tasks
-          .toIndexedSeq
-
-        indices.map { index =>
-          fn(quantileTasks(index.toInt)).toDouble
-        }.toIndexedSeq
-      } else {
-        Utils.tryWithResource(
-          store.view(classOf[TaskDataWrapper])
-            .parent(stageKey)
-            .index(index)
-            .first(0L)
-            .closeableIterator()
-        ) { it =>
-          var last = Double.NaN
-          var currentIdx = -1L
-          indices.map { idx =>
-            if (idx == currentIdx) {
+          .closeableIterator()
+      ) { it =>
+        var last = Double.NaN
+        var currentIdx = -1L
+        indices.map { idx =>
+          if (idx == currentIdx) {
+            last
+          } else {
+            val diff = idx - currentIdx
+            currentIdx = idx
+            if (it.skip(diff - 1)) {
+              last = fn(it.next()).toDouble
               last
             } else {
-              val diff = idx - currentIdx
-              currentIdx = idx
-              if (it.skip(diff - 1)) {
-                last = fn(it.next()).toDouble
-                last
-              } else {
-                Double.NaN
-              }
+              Double.NaN
             }
-          }.toIndexedSeq
-        }
+          }
+        }.toIndexedSeq
       }
     }
 
     val computedQuantiles = new v1.TaskMetricDistributions(
       quantiles = quantiles,
-      executorDeserializeTime = scanTasks(TaskIndexNames.DESER_TIME) { t =>
+      executorDeserializeTime = scanTasks(SuccessTaskIndexNames.DESER_TIME) { t =>
         t.executorDeserializeTime
       },
-      executorDeserializeCpuTime = scanTasks(TaskIndexNames.DESER_CPU_TIME) { t =>
+      executorDeserializeCpuTime = scanTasks(SuccessTaskIndexNames.DESER_CPU_TIME) { t =>
         t.executorDeserializeCpuTime
       },
-      executorRunTime = scanTasks(TaskIndexNames.EXEC_RUN_TIME) { t => t.executorRunTime },
-      executorCpuTime = scanTasks(TaskIndexNames.EXEC_CPU_TIME) { t => t.executorCpuTime },
-      resultSize = scanTasks(TaskIndexNames.RESULT_SIZE) { t => t.resultSize },
-      jvmGcTime = scanTasks(TaskIndexNames.GC_TIME) { t => t.jvmGcTime },
-      resultSerializationTime = scanTasks(TaskIndexNames.SER_TIME) { t =>
+      executorRunTime = scanTasks(SuccessTaskIndexNames.EXEC_RUN_TIME) { t => t.executorRunTime },
+      executorCpuTime = scanTasks(SuccessTaskIndexNames.EXEC_CPU_TIME) { t => t.executorCpuTime },
+      resultSize = scanTasks(SuccessTaskIndexNames.RESULT_SIZE) { t => t.resultSize },
+      jvmGcTime = scanTasks(SuccessTaskIndexNames.GC_TIME) { t => t.jvmGcTime },
+      resultSerializationTime = scanTasks(SuccessTaskIndexNames.SER_TIME) { t =>
         t.resultSerializationTime
       },
-      gettingResultTime = scanTasks(TaskIndexNames.GETTING_RESULT_TIME) { t =>
+      gettingResultTime = scanTasks(SuccessTaskIndexNames.GETTING_RESULT_TIME) { t =>
         t.gettingResultTime
       },
-      schedulerDelay = scanTasks(TaskIndexNames.SCHEDULER_DELAY) { t => t.schedulerDelay },
-      peakExecutionMemory = scanTasks(TaskIndexNames.PEAK_MEM) { t => t.peakExecutionMemory },
-      memoryBytesSpilled = scanTasks(TaskIndexNames.MEM_SPILL) { t => t.memoryBytesSpilled },
-      diskBytesSpilled = scanTasks(TaskIndexNames.DISK_SPILL) { t => t.diskBytesSpilled },
+      schedulerDelay = scanTasks(SuccessTaskIndexNames.SCHEDULER_DELAY) { t => t.schedulerDelay },
+      peakExecutionMemory = scanTasks(SuccessTaskIndexNames.PEAK_MEM) { t =>
+        t.peakExecutionMemory },
+      memoryBytesSpilled = scanTasks(SuccessTaskIndexNames.MEM_SPILL) { t => t.memoryBytesSpilled },
+      diskBytesSpilled = scanTasks(SuccessTaskIndexNames.DISK_SPILL) { t => t.diskBytesSpilled },
       inputMetrics = new v1.InputMetricDistributions(
-        scanTasks(TaskIndexNames.INPUT_SIZE) { t => t.inputBytesRead },
-        scanTasks(TaskIndexNames.INPUT_RECORDS) { t => t.inputRecordsRead }),
+        scanTasks(SuccessTaskIndexNames.INPUT_SIZE) { t => t.inputBytesRead },
+        scanTasks(SuccessTaskIndexNames.INPUT_RECORDS) { t => t.inputRecordsRead }),
       outputMetrics = new v1.OutputMetricDistributions(
-        scanTasks(TaskIndexNames.OUTPUT_SIZE) { t => t.outputBytesWritten },
-        scanTasks(TaskIndexNames.OUTPUT_RECORDS) { t => t.outputRecordsWritten }),
+        scanTasks(SuccessTaskIndexNames.OUTPUT_SIZE) { t => t.outputBytesWritten },
+        scanTasks(SuccessTaskIndexNames.OUTPUT_RECORDS) { t => t.outputRecordsWritten }),
       shuffleReadMetrics = new v1.ShuffleReadMetricDistributions(
-        scanTasks(TaskIndexNames.SHUFFLE_TOTAL_READS) { m =>
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_TOTAL_READS) { m =>
           m.shuffleLocalBytesRead + m.shuffleRemoteBytesRead
         },
-        scanTasks(TaskIndexNames.SHUFFLE_READ_RECORDS) { t => t.shuffleRecordsRead },
-        scanTasks(TaskIndexNames.SHUFFLE_REMOTE_BLOCKS) { t => t.shuffleRemoteBlocksFetched },
-        scanTasks(TaskIndexNames.SHUFFLE_LOCAL_BLOCKS) { t => t.shuffleLocalBlocksFetched },
-        scanTasks(TaskIndexNames.SHUFFLE_READ_TIME) { t => t.shuffleFetchWaitTime },
-        scanTasks(TaskIndexNames.SHUFFLE_REMOTE_READS) { t => t.shuffleRemoteBytesRead },
-        scanTasks(TaskIndexNames.SHUFFLE_REMOTE_READS_TO_DISK) { t =>
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_READ_RECORDS) { t => t.shuffleRecordsRead },
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_REMOTE_BLOCKS) { t =>
+          t.shuffleRemoteBlocksFetched },
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_LOCAL_BLOCKS) { t => t.shuffleLocalBlocksFetched },
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_READ_TIME) { t => t.shuffleFetchWaitTime },
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_REMOTE_READS) { t => t.shuffleRemoteBytesRead },
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_REMOTE_READS_TO_DISK) { t =>
           t.shuffleRemoteBytesReadToDisk
         },
-        scanTasks(TaskIndexNames.SHUFFLE_TOTAL_BLOCKS) { m =>
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_TOTAL_BLOCKS) { m =>
           m.shuffleLocalBlocksFetched + m.shuffleRemoteBlocksFetched
         }),
       shuffleWriteMetrics = new v1.ShuffleWriteMetricDistributions(
-        scanTasks(TaskIndexNames.SHUFFLE_WRITE_SIZE) { t => t.shuffleBytesWritten },
-        scanTasks(TaskIndexNames.SHUFFLE_WRITE_RECORDS) { t => t.shuffleRecordsWritten },
-        scanTasks(TaskIndexNames.SHUFFLE_WRITE_TIME) { t => t.shuffleWriteTime }))
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_WRITE_SIZE) { t => t.shuffleBytesWritten },
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_WRITE_RECORDS) { t => t.shuffleRecordsWritten },
+        scanTasks(SuccessTaskIndexNames.SHUFFLE_WRITE_TIME) { t => t.shuffleWriteTime }))
 
     // Go through the computed quantiles and cache the values that match the caching criteria.
     computedQuantiles.quantiles.zipWithIndex

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

@@ -97,6 +97,41 @@ private[spark] class StageDataWrapper(
   private def completionTime: Long = info.completionTime.map(_.getTime).getOrElse(-1L)
 }
 
+/**
+ * This object map the indices names of successful tasks' metrices. Mapped to short strings
+ * to save space when using a disk store.
+ */
+private[spark] object SuccessTaskIndexNames {
+  final val DESER_CPU_TIME = "sdct"
+  final val DESER_TIME = "sdes"
+  final val DISK_SPILL = "sdbs"
+  final val EXEC_CPU_TIME = "sect"
+  final val EXEC_RUN_TIME = "sert"
+  final val GC_TIME = "sgc"
+  final val GETTING_RESULT_TIME = "sgrt"
+  final val INPUT_RECORDS = "sir"
+  final val INPUT_SIZE = "sis"
+  final val MEM_SPILL = "smbs"
+  final val OUTPUT_RECORDS = "sor"
+  final val OUTPUT_SIZE = "sos"
+  final val PEAK_MEM = "spem"
+  final val RESULT_SIZE = "srs"
+  final val SCHEDULER_DELAY = "sdly"
+  final val SER_TIME = "srst"
+  final val SHUFFLE_LOCAL_BLOCKS = "sslbl"
+  final val SHUFFLE_READ_RECORDS = "ssrr"
+  final val SHUFFLE_READ_TIME = "ssrt"
+  final val SHUFFLE_REMOTE_BLOCKS = "ssrbl"
+  final val SHUFFLE_REMOTE_READS = "ssrby"
+  final val SHUFFLE_REMOTE_READS_TO_DISK = "ssrbd"
+  final val SHUFFLE_TOTAL_READS = "sstby"
+  final val SHUFFLE_TOTAL_BLOCKS = "sstbl"
+  final val SHUFFLE_WRITE_RECORDS = "sswr"
+  final val SHUFFLE_WRITE_SIZE = "ssws"
+  final val SHUFFLE_WRITE_TIME = "sswt"
+  final val STAGE = "stage"
+}
+
 /**
  * Tasks have a lot of indices that are used in a few different places. This object keeps logical
  * names for these indices, mapped to short strings to save space when using a disk store.
@@ -235,6 +270,8 @@ private[spark] class TaskDataWrapper(
 
   def hasMetrics: Boolean = executorDeserializeTime >= 0
 
+  private val isSuccess = status == "SUCCESS"
+
   def toApi: TaskData = {
     val metrics = if (hasMetrics) {
       Some(new TaskMetrics(
@@ -290,6 +327,119 @@ private[spark] class TaskDataWrapper(
       gettingResultTime = 0L)
   }
 
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.DESER_TIME, parent = TaskIndexNames.STAGE)
+  def executorDeserializeTimeIndex: Long = if (isSuccess) {
+    executorDeserializeTime
+  } else {
+    -1L
+  }
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.DESER_CPU_TIME, parent = TaskIndexNames.STAGE)
+  def executorDeserializeCpuTimeIndex: Long = if (isSuccess) {
+    executorDeserializeCpuTime
+  } else {
+    -1L
+  }
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.EXEC_RUN_TIME, parent = TaskIndexNames.STAGE)
+  def executorRunTimeIndex: Long = if (isSuccess) executorRunTime else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.EXEC_CPU_TIME, parent = TaskIndexNames.STAGE)
+  def executorCpuTimeIndex: Long = if (isSuccess) executorCpuTime else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.RESULT_SIZE, parent = TaskIndexNames.STAGE)
+  def resultSizeIndex: Long = if (isSuccess) resultSize else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.GC_TIME, parent = TaskIndexNames.STAGE)
+  def jvmGcTimeIndex: Long = if (isSuccess) jvmGcTime else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SER_TIME, parent = TaskIndexNames.STAGE)
+  def resultSerializationTimeIndex: Long = if (isSuccess) resultSerializationTime else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.MEM_SPILL, parent = TaskIndexNames.STAGE)
+  def memoryBytesSpilledIndex: Long = if (isSuccess) memoryBytesSpilled else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.DISK_SPILL, parent = TaskIndexNames.STAGE)
+  def diskBytesSpilledIndex: Long = if (isSuccess) diskBytesSpilled else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.PEAK_MEM, parent = TaskIndexNames.STAGE)
+  def peakExecutionMemoryIndex: Long = if (isSuccess) peakExecutionMemory else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.INPUT_SIZE, parent = TaskIndexNames.STAGE)
+  def inputBytesReadIndex: Long = if (isSuccess) inputBytesRead else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.INPUT_RECORDS, parent = TaskIndexNames.STAGE)
+  def inputRecordsReadIndex: Long = if (isSuccess) inputRecordsRead else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.OUTPUT_SIZE, parent = TaskIndexNames.STAGE)
+  def outputBytesWrittenIndex: Long = if (isSuccess) outputBytesWritten else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.OUTPUT_RECORDS, parent = TaskIndexNames.STAGE)
+  def outputRecordsWrittenIndex: Long = if (isSuccess) outputRecordsWritten else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_REMOTE_BLOCKS,
+    parent = TaskIndexNames.STAGE)
+  def shuffleRemoteBlocksFetchedIndex: Long = if (isSuccess) {
+    shuffleRemoteBlocksFetched
+  } else {
+    -1L
+  }
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_LOCAL_BLOCKS,
+    parent = TaskIndexNames.STAGE)
+  def shuffleLocalBlocksFetchedIndex: Long = if (isSuccess) {
+    shuffleLocalBlocksFetched
+  } else {
+    -1L
+  }
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_READ_TIME,
+    parent = TaskIndexNames.STAGE)
+  def shuffleFetchWaitTimeIndex: Long = if (isSuccess) shuffleFetchWaitTime else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_REMOTE_READS,
+    parent = TaskIndexNames.STAGE)
+  def shuffleRemoteBytesReadIndex: Long = if (isSuccess) shuffleRemoteBytesRead else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_REMOTE_READS_TO_DISK,
+    parent = TaskIndexNames.STAGE)
+  def shuffleRemoteBytesReadToDiskIndex: Long = if (isSuccess) {
+    shuffleRemoteBytesReadToDisk
+  } else {
+    -1L
+  }
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_READ_RECORDS,
+    parent = TaskIndexNames.STAGE)
+  def shuffleRecordsReadIndex: Long = if (isSuccess) shuffleRecordsRead else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_WRITE_SIZE,
+    parent = TaskIndexNames.STAGE)
+  def shuffleBytesWrittenIndex: Long = if (isSuccess) shuffleBytesWritten else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_WRITE_TIME,
+    parent = TaskIndexNames.STAGE)
+  def shuffleWriteTimeIndex: Long = if (isSuccess) shuffleWriteTime else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_WRITE_RECORDS,
+    parent = TaskIndexNames.STAGE)
+  def shuffleRecordsWrittenIndex: Long = if (isSuccess) shuffleRecordsWritten else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SCHEDULER_DELAY, parent = TaskIndexNames.STAGE)
+  def schedulerDelayIndex: Long = if (isSuccess) schedulerDelay else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.GETTING_RESULT_TIME,
+    parent = TaskIndexNames.STAGE)
+  def gettingResultTimeIndex: Long = if (isSuccess) gettingResultTime else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_TOTAL_READS,
+    parent = TaskIndexNames.STAGE)
+  def shuffleTotalReadsIndex: Long = if (isSuccess) shuffleTotalReads else -1L
+
+  @JsonIgnore @KVIndex(value = SuccessTaskIndexNames.SHUFFLE_TOTAL_BLOCKS,
+    parent = TaskIndexNames.STAGE)
+  private def shuffleTotalBlocksIndex: Long = if (isSuccess) shuffleTotalBlocks else -1L
+
   @JsonIgnore @KVIndex(TaskIndexNames.STAGE)
   private def stage: Array[Int] = Array(stageId, stageAttemptId)