Handle Double.Nan in json

apache · Nov 29, 2016 · f6d60df · f6d60df
1 parent 27e3b73
commit f6d60df
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 22 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala
@@ -119,23 +119,13 @@ trait ProgressReporter extends Logging {
 
     val sourceProgress = sources.map { source =>
       val numRecords = executionStats.inputRows.getOrElse(source, 0L)
-      val inputRecordsPerSecond = if (inputTimeSec > 0) {
-        numRecords / inputTimeSec
-      } else {
-        Double.MaxValue
-      }
-      val processedRecordsPerSecond = if (processingTimeSec > 0) {
-        numRecords / processingTimeSec
-      } else {
-        Double.MaxValue
-      }
       new SourceProgress(
         description = source.toString,
         startOffset = committedOffsets.get(source).map(_.json).orNull,
         endOffset = availableOffsets.get(source).map(_.json).orNull,
         numInputRows = numRecords,
-        inputRowsPerSecond = inputRecordsPerSecond,
-        processedRowsPerSecond = processedRecordsPerSecond
+        inputRowsPerSecond = numRecords / inputTimeSec,
+        processedRowsPerSecond = numRecords / processingTimeSec
       )
     }
 

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/SourceProgress.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/SourceProgress.scala
@@ -57,12 +57,16 @@ class SourceProgress protected[sql](
   override def toString: String = prettyJson
 
   private[sql] def jsonValue: JValue = {
+    def safeDoubleToJValue(value: Double): JValue = {
+      if (value.isNaN || value.isInfinity) JNothing else JDouble(value)
+    }
+
     ("description" -> JString(description)) ~
     ("startOffset" -> tryParse(startOffset)) ~
     ("endOffset" -> tryParse(endOffset)) ~
     ("numInputRows" -> JInt(numInputRows)) ~
-    ("inputRowsPerSecond" -> JDouble(inputRowsPerSecond)) ~
-    ("processedRowsPerSecond" -> JDouble(processedRowsPerSecond))
+    ("inputRowsPerSecond" -> safeDoubleToJValue(inputRowsPerSecond)) ~
+    ("processedRowsPerSecond" -> safeDoubleToJValue(processedRowsPerSecond))
   }
 
   private def tryParse(json: String) = try {

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryProgress.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryProgress.scala
@@ -92,12 +92,16 @@ class StreamingQueryProgress private[sql](
   override def toString: String = prettyJson
 
   private[sql] def jsonValue: JValue = {
+    def safeDoubleToJValue(value: Double): JValue = {
+      if (value.isNaN || value.isInfinity) JNothing else JDouble(value)
+    }
+
     ("id" -> JString(id.toString)) ~
     ("name" -> JString(name)) ~
     ("timestamp" -> JInt(timestamp)) ~
     ("numInputRows" -> JInt(numInputRows)) ~
-    ("inputRowsPerSecond" -> JDouble(inputRowsPerSecond)) ~
-    ("processedRowsPerSecond" -> JDouble(processedRowsPerSecond)) ~
+    ("inputRowsPerSecond" -> safeDoubleToJValue(inputRowsPerSecond)) ~
+    ("processedRowsPerSecond" -> safeDoubleToJValue(processedRowsPerSecond)) ~
     ("durationMs" -> durationMs
         .asScala
         .map { case (k, v) => k -> JInt(v.toLong): JObject }

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryProgressSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryProgressSuite.scala
@@ -40,7 +40,6 @@ class StreamingQueryProgressSuite extends SparkFunSuite {
         |  "timestamp" : 1,
         |  "numInputRows" : 678,
         |  "inputRowsPerSecond" : 10.0,
-        |  "processedRowsPerSecond" : 1.7976931348623157E308,
         |  "durationMs" : {
         |    "total" : 0
         |  },
@@ -54,17 +53,16 @@ class StreamingQueryProgressSuite extends SparkFunSuite {
         |    "startOffset" : 123,
         |    "endOffset" : 456,
         |    "numInputRows" : 678,
-        |    "inputRowsPerSecond" : 10.0,
-        |    "processedRowsPerSecond" : 1.7976931348623157E308
+        |    "inputRowsPerSecond" : 10.0
         |  } ]
         |}
       """.stripMargin.trim)
-    assert(parse(json) === testProgress.jsonValue)
+    assert(compact(parse(json)) === testProgress.json)
 
   }
 
   test("json") {
-    assert(parse(testProgress.json) === testProgress.jsonValue)
+    assert(compact(parse(testProgress.json)) === testProgress.json)
   }
 
   test("toString") {
@@ -88,7 +86,7 @@ object StreamingQueryProgressSuite {
         endOffset = "456",
         numInputRows = 678,
         inputRowsPerSecond = 10.0,
-        processedRowsPerSecond = Double.MaxValue
+        processedRowsPerSecond = Double.PositiveInfinity  // should not be present in the json
       )
     )
   )