Refactored

Author: tdas

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

@@ -157,10 +157,10 @@ trait ProgressReporter extends Logging {
       id = id,
       runId = runId,
       name = name,
-      triggerTimestamp = formatTimestamp(currentTriggerStartTimestamp),
+      timestamp = formatTimestamp(currentTriggerStartTimestamp),
       batchId = currentBatchId,
       durationMs = currentDurationsMs.toMap.mapValues(long2Long).asJava,
-      queryTimestamps = executionStats.eventTimeStats.asJava,
+      eventTime = executionStats.eventTimeStats.asJava,
       stateOperators = executionStats.stateOperators.toArray,
       sources = sourceProgress.toArray,
       sink = sinkProgress)
@@ -182,12 +182,10 @@ trait ProgressReporter extends Logging {
 
   /** Extracts statistics from the most recent query execution. */
   private def extractExecutionStats(hasNewData: Boolean): ExecutionStats = {
-    val basicQueryTimestamps = Map(
-      "watermark" -> offsetSeqMetadata.batchWatermarkMs,
-      "processingTime" -> offsetSeqMetadata.batchTimestampMs).mapValues(formatTimestamp)
+    val watermarkTimestamp = Map("watermark" -> formatTimestamp(offsetSeqMetadata.batchWatermarkMs))
 
     if (!hasNewData) {
-      return ExecutionStats(Map.empty, Seq.empty, basicQueryTimestamps)
+      return ExecutionStats(Map.empty, Seq.empty, watermarkTimestamp)
     }
 
     // We want to associate execution plan leaves to sources that generate them, so that we match
@@ -250,10 +248,10 @@ trait ProgressReporter extends Logging {
       case e: EventTimeWatermarkExec if e.eventTimeStats.value.count > 0 =>
         val stats = e.eventTimeStats.value
         Map(
-          "eventTime.max" -> stats.max,
-          "eventTime.min" -> stats.min,
-          "eventTime.avg" -> stats.avg).mapValues(formatTimestamp)
-    }.headOption.getOrElse(Map.empty) ++ basicQueryTimestamps
+          "max" -> stats.max,
+          "min" -> stats.min,
+          "avg" -> stats.avg).mapValues(formatTimestamp)
+    }.headOption.getOrElse(Map.empty) ++ watermarkTimestamp
 
     ExecutionStats(numInputRows, stateOperators, eventTimeStats)
   }

sql/core/src/main/scala/org/apache/spark/sql/streaming/progress.scala

@@ -41,13 +41,13 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
  * @param id An unique query id that persists across restarts. See `StreamingQuery.id()`.
  * @param runId A query id that is unique for every start/restart. See `StreamingQuery.runId()`.
  * @param name User-specified name of the query, null if not specified.
- * @param triggerTimestamp Timestamp (ms) of the beginning of the trigger.
+ * @param timestamp Timestamp (ms) of the beginning of the trigger.
  * @param batchId A unique id for the current batch of data being processed.  Note that in the
  *                case of retries after a failure a given batchId my be executed more than once.
  *                Similarly, when there is no data to be processed, the batchId will not be
  *                incremented.
  * @param durationMs The amount of time taken to perform various operations in milliseconds.
- * @param queryTimestamps Statistics of event time seen in this batch
+ * @param eventTime Statistics of event time seen in this batch
  * @param stateOperators Information about operators in the query that store state.
  * @param sources detailed statistics on data being read from each of the streaming sources.
  * @since 2.1.0
@@ -57,10 +57,10 @@ class StreamingQueryProgress private[sql](
   val id: UUID,
   val runId: UUID,
   val name: String,
-  val triggerTimestamp: String,
+  val timestamp: String,
   val batchId: Long,
   val durationMs: ju.Map[String, JLong],
-  val queryTimestamps: ju.Map[String, String],
+  val eventTime: ju.Map[String, String],
   val stateOperators: Array[StateOperatorProgress],
   val sources: Array[SourceProgress],
   val sink: SinkProgress) {
@@ -98,12 +98,12 @@ class StreamingQueryProgress private[sql](
     ("id" -> JString(id.toString)) ~
     ("runId" -> JString(runId.toString)) ~
     ("name" -> JString(name)) ~
-    ("triggerTimestamp" -> JString(triggerTimestamp)) ~
+    ("timestamp" -> JString(timestamp)) ~
     ("numInputRows" -> JInt(numInputRows)) ~
     ("inputRowsPerSecond" -> safeDoubleToJValue(inputRowsPerSecond)) ~
     ("processedRowsPerSecond" -> safeDoubleToJValue(processedRowsPerSecond)) ~
     ("durationMs" -> safeMapToJValue[JLong](durationMs, v => JInt(v.toLong))) ~
-    ("queryTimestamps" -> safeMapToJValue[String](queryTimestamps, s => JString(s))) ~
+    ("eventTime" -> safeMapToJValue[String](eventTime, s => JString(s))) ~
     ("stateOperators" -> JArray(stateOperators.map(_.jsonValue).toList)) ~
     ("sources" -> JArray(sources.map(_.jsonValue).toList)) ~
     ("sink" -> sink.jsonValue)

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

@@ -190,7 +190,7 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter {
       val newEvent = JsonProtocol.sparkEventFromJson(json).asInstanceOf[QueryProgressEvent]
       assert(newEvent.progress.json === event.progress.json)  // json as a proxy for equality
       assert(newEvent.progress.durationMs.asScala === event.progress.durationMs.asScala)
-      assert(newEvent.progress.queryTimestamps.asScala === event.progress.queryTimestamps.asScala)
+      assert(newEvent.progress.eventTime.asScala === event.progress.eventTime.asScala)
     }
     testSerialization(new QueryProgressEvent(StreamingQueryStatusAndProgressSuite.testProgress1))
     testSerialization(new QueryProgressEvent(StreamingQueryStatusAndProgressSuite.testProgress2))

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

@@ -38,17 +38,17 @@ class StreamingQueryStatusAndProgressSuite extends SparkFunSuite {
         |  "id" : "${testProgress1.id.toString}",
         |  "runId" : "${testProgress1.runId.toString}",
         |  "name" : "myName",
-        |  "triggerTimestamp" : "2016-12-05T20:54:20.827Z",
+        |  "timestamp" : "2016-12-05T20:54:20.827Z",
         |  "numInputRows" : 678,
         |  "inputRowsPerSecond" : 10.0,
         |  "durationMs" : {
         |    "total" : 0
         |  },
-        |  "queryTimestamps" : {
-        |    "eventTime.avg" : "2016-12-05T20:54:20.827Z",
-        |    "eventTime.max" : "2016-12-05T20:54:20.827Z",
-        |    "eventTime.min" : "2016-12-05T20:54:20.827Z",
-        |    "processingTime" : "2016-12-05T20:54:20.827Z"
+        |  "eventTime" : {
+        |    "avg" : "2016-12-05T20:54:20.827Z",
+        |    "max" : "2016-12-05T20:54:20.827Z",
+        |    "min" : "2016-12-05T20:54:20.827Z",
+        |    "watermark" : "2016-12-05T20:54:20.827Z"
         |  },
         |  "stateOperators" : [ {
         |    "numRowsTotal" : 0,
@@ -76,7 +76,7 @@ class StreamingQueryStatusAndProgressSuite extends SparkFunSuite {
          |  "id" : "${testProgress2.id.toString}",
          |  "runId" : "${testProgress2.runId.toString}",
          |  "name" : null,
-         |  "triggerTimestamp" : "2016-12-05T20:54:20.827Z",
+         |  "timestamp" : "2016-12-05T20:54:20.827Z",
          |  "numInputRows" : 678,
          |  "durationMs" : {
          |    "total" : 0
@@ -135,14 +135,14 @@ object StreamingQueryStatusAndProgressSuite {
     id = UUID.randomUUID,
     runId = UUID.randomUUID,
     name = "myName",
-    triggerTimestamp = "2016-12-05T20:54:20.827Z",
+    timestamp = "2016-12-05T20:54:20.827Z",
     batchId = 2L,
     durationMs = Map("total" -> 0L).mapValues(long2Long).asJava,
-    queryTimestamps = Map(
-      "eventTime.max" -> "2016-12-05T20:54:20.827Z",
-      "eventTime.min" -> "2016-12-05T20:54:20.827Z",
-      "eventTime.avg" -> "2016-12-05T20:54:20.827Z",
-      "processingTime" -> "2016-12-05T20:54:20.827Z").asJava,
+    eventTime = Map(
+      "max" -> "2016-12-05T20:54:20.827Z",
+      "min" -> "2016-12-05T20:54:20.827Z",
+      "avg" -> "2016-12-05T20:54:20.827Z",
+      "watermark" -> "2016-12-05T20:54:20.827Z").asJava,
     stateOperators = Array(new StateOperatorProgress(numRowsTotal = 0, numRowsUpdated = 1)),
     sources = Array(
       new SourceProgress(
@@ -161,10 +161,10 @@ object StreamingQueryStatusAndProgressSuite {
     id = UUID.randomUUID,
     runId = UUID.randomUUID,
     name = null, // should not be present in the json
-    triggerTimestamp = "2016-12-05T20:54:20.827Z",
+    timestamp = "2016-12-05T20:54:20.827Z",
     batchId = 2L,
     durationMs = Map("total" -> 0L).mapValues(long2Long).asJava,
-    queryTimestamps = Map.empty[String, String].asJava,  // empty maps should be handled correctly
+    eventTime = Map.empty[String, String].asJava,  // empty maps should be handled correctly
     stateOperators = Array(new StateOperatorProgress(numRowsTotal = 0, numRowsUpdated = 1)),
     sources = Array(
       new SourceProgress(

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

@@ -245,7 +245,7 @@ class StreamingQuerySuite extends StreamTest with BeforeAndAfter with Logging {
         assert(progress.id === query.id)
         assert(progress.name === query.name)
         assert(progress.batchId === 0)
-        assert(progress.triggerTimestamp === "1970-01-01T00:00:00.100Z") // 100 ms in UTC
+        assert(progress.timestamp === "1970-01-01T00:00:00.100Z") // 100 ms in UTC
         assert(progress.numInputRows === 2)
         assert(progress.processedRowsPerSecond === 2.0)
 
@@ -255,8 +255,6 @@ class StreamingQuerySuite extends StreamTest with BeforeAndAfter with Logging {
         assert(progress.durationMs.get("walCommit") === 0)
         assert(progress.durationMs.get("triggerExecution") === 1000)
 
-        assert(progress.queryTimestamps.get("processingTime") === "1970-01-01T00:00:00.300Z")
-        assert(!progress.queryTimestamps.keySet.asScala.exists(_.toLowerCase.contains("eventtime")))
         assert(progress.sources.length === 1)
         assert(progress.sources(0).description contains "MemoryStream")
         assert(progress.sources(0).startOffset === null)

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

@@ -64,41 +64,41 @@ class WatermarkSuite extends StreamTest with BeforeAndAfter with Logging {
         .select($"window".getField("start").cast("long").as[Long], $"count".as[Long])
 
     def assertEventStats(body: ju.Map[String, String] => Unit): AssertOnQuery = AssertOnQuery { q =>
-      body(q.recentProgress.filter(_.numInputRows > 0).lastOption.get.queryTimestamps)
+      body(q.recentProgress.filter(_.numInputRows > 0).lastOption.get.eventTime)
       true
     }
 
     testStream(windowedAggregation)(
       AddData(inputData, 15),
       CheckAnswer(),
       assertEventStats { e =>
-        assert(e.get("eventTime.max") === formatTimestamp(15))
-        assert(e.get("eventTime.min") === formatTimestamp(15))
-        assert(e.get("eventTime.avg") === formatTimestamp(15))
+        assert(e.get("max") === formatTimestamp(15))
+        assert(e.get("min") === formatTimestamp(15))
+        assert(e.get("avg") === formatTimestamp(15))
         assert(e.get("watermark") === formatTimestamp(0))
       },
       AddData(inputData, 10, 12, 14),
       CheckAnswer(),
       assertEventStats { e =>
-        assert(e.get("eventTime.max") === formatTimestamp(14))
-        assert(e.get("eventTime.min") === formatTimestamp(10))
-        assert(e.get("eventTime.avg") === formatTimestamp(12))
+        assert(e.get("max") === formatTimestamp(14))
+        assert(e.get("min") === formatTimestamp(10))
+        assert(e.get("avg") === formatTimestamp(12))
         assert(e.get("watermark") === formatTimestamp(5))
       },
       AddData(inputData, 25),
       CheckAnswer(),
       assertEventStats { e =>
-        assert(e.get("eventTime.max") === formatTimestamp(25))
-        assert(e.get("eventTime.min") === formatTimestamp(25))
-        assert(e.get("eventTime.avg") === formatTimestamp(25))
+        assert(e.get("max") === formatTimestamp(25))
+        assert(e.get("min") === formatTimestamp(25))
+        assert(e.get("avg") === formatTimestamp(25))
         assert(e.get("watermark") === formatTimestamp(5))
       },
       AddData(inputData, 25),
       CheckAnswer((10, 3)),
       assertEventStats { e =>
-        assert(e.get("eventTime.max") === formatTimestamp(25))
-        assert(e.get("eventTime.min") === formatTimestamp(25))
-        assert(e.get("eventTime.avg") === formatTimestamp(25))
+        assert(e.get("max") === formatTimestamp(25))
+        assert(e.get("min") === formatTimestamp(25))
+        assert(e.get("avg") === formatTimestamp(25))
         assert(e.get("watermark") === formatTimestamp(15))
       }
     )