Fix duration minimum for zero-task stages

core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/util/TaskMetricsAccumRec.scala

@@ -142,8 +142,12 @@ class TaskMetricsAccumRec {
     inputBytesReadMax = math.max(inputBytesReadMax, rec.inputBytesReadMax)
     peakExecutionMemoryMax = math.max(peakExecutionMemoryMax, rec.peakExecutionMemoryMax)
     resultSizeMax = math.max(resultSizeMax, rec.resultSizeMax)
-    // Min
-    durationMin = math.min(durationMin, rec.durationMin)
+    // Min. A record with no tasks carries durationMin = 0 from resetFields, which is a
+    // placeholder rather than a measurement. durationMin seeds at Long.MaxValue, the identity
+    // for min, so an empty accumulator needs no separate case here.
+    if (rec.numTasks > 0) {
+      durationMin = math.min(durationMin, rec.durationMin)
+    }
   }
 
   /**

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala

@@ -836,14 +836,21 @@ trait BaseJobStageAggTaskMetricsProfileResult extends ProfileResult {
     case None => "null"
   }
 
+  /** The minimum is undefined on a record that observed no tasks. */
+  private def durationMinOpt: Option[Long] =
+    if (numTasks > 0) Some(durationMin) else None
+
+  private def durationMinStr: String =
+    StringUtils.optionToString(durationMinOpt, (value: Long) => value.toString)
+
   override def convertToSeq(): Array[String] = {
     Array(id.toString,
       numTasks.toString,
       durStr,
       diskBytesSpilledSum.toString,
       durationSum.toString,
       durationMax.toString,
-      durationMin.toString,
+      durationMinStr,
       durationAvg.toString,
       executorCPUTimeSum.toString,
       executorDeserializeCpuTimeSum.toString,
@@ -950,6 +957,23 @@ case class StageAggTaskMetricsProfileResult(
     swWriteTimeSum: Long // milliseconds
   ) extends BaseJobStageAggTaskMetricsProfileResult {
 
+  /**
+   * Minimum task duration across two attempts, ignoring an attempt that recorded no tasks.
+   * An empty attempt carries durationMin = 0 from TaskMetricsAccumRec.resetFields, which is a
+   * placeholder rather than a measurement, so folding it with Math.min would report zero.
+   */
+  private def minDurationWith(other: StageAggTaskMetricsProfileResult): Long = {
+    if (this.numTasks > 0 && other.numTasks > 0) {
+      Math.min(this.durationMin, other.durationMin)
+    } else if (this.numTasks > 0) {
+      this.durationMin
+    } else if (other.numTasks > 0) {
+      other.durationMin
+    } else {
+      0L
+    }
+  }
+
   /**
    * Combines two StageAggTaskMetricsProfileResults for the same stage.
    * This method aggregates the metrics from the current instance and the provided `other` instance.
@@ -977,7 +1001,7 @@ case class StageAggTaskMetricsProfileResult(
       diskBytesSpilledSum = this.diskBytesSpilledSum + other.diskBytesSpilledSum,
       durationSum = mergedDurationSum,
       durationMax = Math.max(this.durationMax, other.durationMax),
-      durationMin = Math.min(this.durationMin, other.durationMin),
+      durationMin = minDurationWith(other),
       durationAvg = ToolUtils.calculateAverage(mergedDurationSum, mergedNumTasks, 1),
       executorCPUTimeSum = this.executorCPUTimeSum + other.executorCPUTimeSum,
       executorDeserializeCpuTimeSum = this.executorDeserializeCpuTimeSum +

core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala

@@ -649,15 +649,34 @@ class AnalysisSuite extends AnyFunSuite {
       "fixture no longer distinguishes the two rollup formulas; the guard is now vacuous")
   }
 
-  test("stage attempt avg pools task durations rather than averaging attempt means") {
+  test("zero-task stages do not lower SQL duration minimum") {
+    val logs = Array(s"$logDir/gpu_oom_eventlog.zstd")
+    val apps = ToolTestUtils.processProfileApps(logs, sparkSession)
+    val agg = RawMetricProfilerView.getAggMetrics(apps.toSeq)
+    val stagesById = agg.stageAggs.map(row => row.id -> row).toMap
+    val durationMinIndex = stagesById(32L).outputHeaders.indexOf("duration_min")
+
+    assert(durationMinIndex >= 0)
+    Seq(32L, 33L, 37L).foreach { stageId =>
+      val emptyStage = stagesById(stageId)
+      assert(emptyStage.numTasks === 0)
+      assert(emptyStage.durationMin === 0L)
+      assert(emptyStage.convertToCSVSeq()(durationMinIndex).isEmpty)
+    }
+    assert(agg.sqlAggs.find(_.sqlId == 24L)
+      .map(row => (row.numTasks, row.durationMin))
+      .contains((353, 3085L)))
+  }
+
+  test("stage attempts aggregate task duration statistics across empty attempts") {
     val firstAttempt = StageAggTaskMetricsProfileResult(
       id = 1L,
       numTasks = 2,
       duration = None,
       diskBytesSpilledSum = 0L,
       durationSum = 800L,
-      durationMax = 0L,
-      durationMin = 0L,
+      durationMax = 500L,
+      durationMin = 300L,
       durationAvg = 400.0,
       executorCPUTimeSum = 0L,
       executorDeserializeCpuTimeSum = 0L,
@@ -686,11 +705,26 @@ class AnalysisSuite extends AnyFunSuite {
     val retryAttempt = firstAttempt.copy(
       numTasks = 1,
       durationSum = 200L,
+      durationMax = 200L,
+      durationMin = 200L,
       durationAvg = 200.0)
 
     val result = firstAttempt.aggregateStageProfileMetric(retryAttempt)
 
+    // Preserve the existing pooled-average contract while adding minimum coverage.
     assert(result.durationAvg === 333.3)
+    assert(result.durationMin === 200L)
+
+    val emptyAttempt = firstAttempt.copy(
+      numTasks = 0,
+      durationSum = 0L,
+      durationMax = 0L,
+      durationMin = 0L,
+      durationAvg = 0.0)
+
+    assert(firstAttempt.aggregateStageProfileMetric(emptyAttempt).durationMin === 300L)
+    assert(emptyAttempt.aggregateStageProfileMetric(firstAttempt).durationMin === 300L)
+    assert(emptyAttempt.aggregateStageProfileMetric(emptyAttempt).durationMin === 0L)
   }
 
   test("dispersion columns are consistent with the row they sit in") {