done

Author: WweiL

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/progress.scala

@@ -26,7 +26,6 @@ import scala.util.control.NonFatal
 
 import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
-import com.fasterxml.jackson.databind.node.ObjectNode
 import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
 import org.json4s._
 import org.json4s.JsonAST.JValue
@@ -204,11 +203,7 @@ private[spark] object StreamingQueryProgress {
   }
 
   private[spark] def jsonString(progress: StreamingQueryProgress): String = {
-    val jsonNode = mapper.valueToTree[ObjectNode](StreamingQueryProgress)
-    jsonNode.put("numInputRows", progress.numInputRows)
-    jsonNode.put("inputRowsPerSecond", progress.inputRowsPerSecond)
-    jsonNode.put("processedRowsPerSecond", progress.processedRowsPerSecond)
-    mapper.writeValueAsString(progress)
+    progress.json
   }
 
   private[spark] def fromJson(json: String): StreamingQueryProgress =

connector/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala

@@ -3259,10 +3259,6 @@ class SparkConnectPlanner(
       case StreamingQueryCommand.CommandCase.LAST_PROGRESS |
           StreamingQueryCommand.CommandCase.RECENT_PROGRESS =>
         val progressReports = if (command.getLastProgress) {
-          // scalastyle:off
-          println("wei== sparkconnnectplanner lasprogress" + query.lastProgress)
-          println("wei== sparkconnnectplanner jsonString" + StreamingQueryProgress.jsonString(query.lastProgress))
-          // scalastyle:on
           Option(query.lastProgress).toSeq
         } else {
           query.recentProgress.toImmutableArraySeq

python/pyspark/sql/tests/streaming/test_streaming.py

@@ -28,7 +28,7 @@
 
 class StreamingTestsMixin:
     def test_streaming_query_functions_basic(self):
-        df = self.spark.readStream.format("rate").option("rowsPerSecond", 10).load()
+        df = self.spark.readStream.format("text").load("python/test_support/sql/streaming")
         query = (
             df.writeStream.format("memory")
             .queryName("test_streaming_query_functions_basic")

python/pyspark/sql/tests/streaming/test_streaming_listener.py

@@ -194,6 +194,17 @@ def check_sink_progress(self, progress):
         self.assertTrue(isinstance(progress.numOutputRows, int))
         self.assertTrue(isinstance(progress.metrics, dict))
 
+    # def test_streaming_last_progress(self):
+    #     try:
+    #         df = self.spark.readStream.format("text").load("python/test_support/sql/streaming")
+    #         query = df.writeStream.format("noop").queryName("test_streaming_progress").start()
+    #         query.processAllAvailable()
+    #
+    #         progress = StreamingQueryProgress.fromJson(query.lastProgress)
+    #         self.check_streaming_query_progress(progress, False)
+    #     finally:
+    #         query.stop()
+
     # This is a generic test work for both classic Spark and Spark Connect
     def test_listener_observed_metrics(self):
         class MyErrorListener(StreamingQueryListener):

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

@@ -26,7 +26,6 @@ import scala.util.control.NonFatal
 
 import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
-import com.fasterxml.jackson.databind.node.ObjectNode
 import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
 import org.json4s._
 import org.json4s.JsonAST.JValue
@@ -188,11 +187,7 @@ private[spark] object StreamingQueryProgress {
   }
 
   private[spark] def jsonString(progress: StreamingQueryProgress): String = {
-    val jsonNode = mapper.valueToTree[ObjectNode](StreamingQueryProgress)
-    jsonNode.put("numInputRows", progress.numInputRows)
-    jsonNode.put("inputRowsPerSecond", progress.inputRowsPerSecond)
-    jsonNode.put("processedRowsPerSecond", progress.processedRowsPerSecond)
-    mapper.writeValueAsString(progress)
+    progress.json
   }
 
   private[spark] def fromJson(json: String): StreamingQueryProgress =