Address

Author: brkyvz

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala

@@ -180,11 +180,7 @@ trait ProgressReporter extends Logging {
     currentStatus = currentStatus.copy(isTriggerActive = false)
   }
 
-  /**
-   * Extract statistics about stateful operators from the executed query plan.
-   * SPARK-19378: Still report stateOperator metrics even though no data was processed while
-   * reporting progress.
-   */
+  /** Extract statistics about stateful operators from the executed query plan. */
   private def extractStateOperatorMetrics(hasNewData: Boolean): Seq[StateOperatorProgress] = {
     if (lastExecution == null) return Nil
     // lastExecution could belong to one of the previous triggers if `!hasNewData`.
@@ -204,32 +200,31 @@ trait ProgressReporter extends Logging {
     }
   }
 
-  /**
-   * Extract statistics about event time from the executed query plan.
-   * SPARK-19378: Still report eventTime metrics even though no data was processed while
-   * reporting progress.
-   */
-  private def extractEventTimeStats(watermarkTs: Map[String, String]): Map[String, String] = {
-    if (lastExecution == null) return watermarkTs
+  /** Extract statistics about event time from the executed query plan. */
+  private def extractEventTimeStats(): Map[String, String] = {
+    val hasEventTime = logicalPlan.collect { case e: EventTimeWatermark => e }.nonEmpty
+    val watermarkTimestamp =
+      if (hasEventTime) Map("watermark" -> formatTimestamp(offsetSeqMetadata.batchWatermarkMs))
+      else Map.empty[String, String]
+
+    if (lastExecution == null || !hasEventTime) return watermarkTimestamp
+
     lastExecution.executedPlan.collect {
       case e: EventTimeWatermarkExec if e.eventTimeStats.value.count > 0 =>
         val stats = e.eventTimeStats.value
         Map(
           "max" -> stats.max,
           "min" -> stats.min,
           "avg" -> stats.avg).mapValues(formatTimestamp)
-    }.headOption.getOrElse(Map.empty) ++ watermarkTs
+    }.headOption.getOrElse(Map.empty)
   }
 
   /** Extracts statistics from the most recent query execution. */
   private def extractExecutionStats(hasNewData: Boolean): ExecutionStats = {
-    val hasEventTime = logicalPlan.collect { case e: EventTimeWatermark => e }.nonEmpty
-    val watermarkTimestamp =
-      if (hasEventTime) Map("watermark" -> formatTimestamp(offsetSeqMetadata.batchWatermarkMs))
-      else Map.empty[String, String]
 
+    // SPARK-19378: Still report metrics even though no data was processed while reporting progress.
     val stateOperators = extractStateOperatorMetrics(hasNewData)
-    val eventTimeStats = extractEventTimeStats(watermarkTimestamp)
+    val eventTimeStats = extractEventTimeStats()
 
     if (!hasNewData) {
       return ExecutionStats(Map.empty, stateOperators, eventTimeStats)

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryStatusAndProgressSuite.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.streaming
 import java.util.UUID
 
 import scala.collection.JavaConverters._
-import scala.language.postfixOps._
+import scala.language.postfixOps
 
 import org.json4s._
 import org.json4s.jackson.JsonMethods._
@@ -202,8 +202,13 @@ class StreamingQueryStatusAndProgressSuite extends StreamTest with Eventually {
         eventually(timeout(1 minute)) {
           val nextProgress = query.lastProgress
           assert(nextProgress.timestamp !== progress.timestamp)
-          assert(progress.eventTime.size() > 1)
-          assert(progress.stateOperators.length > 0)
+          assert(nextProgress.numInputRows === 0)
+          assert(nextProgress.eventTime.get("min") === "2017-01-26 01:00:00")
+          assert(nextProgress.eventTime.get("avg") === "2017-01-26 01:00:01")
+          assert(nextProgress.eventTime.get("max") === "2017-01-26 01:00:02")
+          assert(nextProgress.stateOperators.head.numRowsTotal === 2)
+          assert(nextProgress.stateOperators.head.numRowsTotal === 2)
+          assert(nextProgress.stateOperators.head.numRowsUpdated === 0)
         }
       } finally {
         query.stop()