[DON'T MERGE] Try to replace all json4s with Jackson

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

@@ -17,12 +17,10 @@
 
 package org.apache.spark.sql.streaming
 
-import org.json4s._
-import org.json4s.JsonAST.JValue
-import org.json4s.JsonDSL._
-import org.json4s.jackson.JsonMethods._
+import com.fasterxml.jackson.databind.JsonNode
 
 import org.apache.spark.annotation.Evolving
+import org.apache.spark.util.JacksonUtils
 
 /**
  * Reports information about the instantaneous status of a streaming query.
@@ -47,10 +45,10 @@ class StreamingQueryStatus protected[sql] (
   // This is a copy of the same class in sql/core/.../streaming/StreamingQueryStatus.scala
 
   /** The compact JSON representation of this status. */
-  def json: String = compact(render(jsonValue))
+  def json: String = JacksonUtils.writeValueAsString(jsonNode)
 
   /** The pretty (i.e. indented) JSON representation of this status. */
-  def prettyJson: String = pretty(render(jsonValue))
+  def prettyJson: String = JacksonUtils.writeValuePrettyAsString(jsonNode)
 
   override def toString: String = prettyJson
 
@@ -64,9 +62,11 @@ class StreamingQueryStatus protected[sql] (
       isTriggerActive = isTriggerActive)
   }
 
-  private[sql] def jsonValue: JValue = {
-    ("message" -> JString(message)) ~
-      ("isDataAvailable" -> JBool(isDataAvailable)) ~
-      ("isTriggerActive" -> JBool(isTriggerActive))
+  private[sql] def jsonNode: JsonNode = {
+    val node = JacksonUtils.createObjectNode
+    node.put("message", message)
+    node.put("isDataAvailable", isDataAvailable)
+    node.put("isTriggerActive", isTriggerActive)
+    node
   }
 }

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

@@ -24,20 +24,17 @@ import java.util.UUID
 import scala.collection.JavaConverters._
 import scala.util.control.NonFatal
 
-import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.databind.{DeserializationFeature, JsonNode, ObjectMapper}
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
 import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
-import org.json4s._
-import org.json4s.JsonAST.JValue
-import org.json4s.JsonDSL._
-import org.json4s.jackson.JsonMethods._
 
 import org.apache.spark.annotation.Evolving
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
-import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJValue, safeMapToJValue}
+import org.apache.spark.sql.streaming.SafeJsonSerializer.{safeDoubleToJsonNode, safeMapToJsonNode}
 import org.apache.spark.sql.streaming.SinkProgress.DEFAULT_NUM_OUTPUT_ROWS
 import org.apache.spark.sql.util.ToJsonUtil
+import org.apache.spark.util.JacksonUtils
 
 /**
  * Information about updates made to stateful operators in a [[StreamingQuery]] during a trigger.
@@ -58,11 +55,12 @@ class StateOperatorProgress private[spark] (
     val customMetrics: ju.Map[String, JLong] = new ju.HashMap())
     extends Serializable {
 
+  private val factory = JacksonUtils.defaultNodeFactory
   /** The compact JSON representation of this progress. */
-  def json: String = compact(render(jsonValue))
+  def json: String = JacksonUtils.writeValueAsString(jsonNode)
 
   /** The pretty (i.e. indented) JSON representation of this progress. */
-  def prettyJson: String = pretty(render(jsonValue))
+  def prettyJson: String = JacksonUtils.writeValuePrettyAsString(jsonNode)
 
   private[sql] def copy(
       newNumRowsUpdated: Long,
@@ -81,26 +79,26 @@ class StateOperatorProgress private[spark] (
       numStateStoreInstances = numStateStoreInstances,
       customMetrics = customMetrics)
 
-  private[sql] def jsonValue: JValue = {
-    ("operatorName" -> JString(operatorName)) ~
-      ("numRowsTotal" -> JInt(numRowsTotal)) ~
-      ("numRowsUpdated" -> JInt(numRowsUpdated)) ~
-      ("allUpdatesTimeMs" -> JInt(allUpdatesTimeMs)) ~
-      ("numRowsRemoved" -> JInt(numRowsRemoved)) ~
-      ("allRemovalsTimeMs" -> JInt(allRemovalsTimeMs)) ~
-      ("commitTimeMs" -> JInt(commitTimeMs)) ~
-      ("memoryUsedBytes" -> JInt(memoryUsedBytes)) ~
-      ("numRowsDroppedByWatermark" -> JInt(numRowsDroppedByWatermark)) ~
-      ("numShufflePartitions" -> JInt(numShufflePartitions)) ~
-      ("numStateStoreInstances" -> JInt(numStateStoreInstances)) ~
-      ("customMetrics" -> {
-        if (!customMetrics.isEmpty) {
-          val keys = customMetrics.keySet.asScala.toSeq.sorted
-          keys.map { k => k -> JInt(customMetrics.get(k).toLong): JObject }.reduce(_ ~ _)
-        } else {
-          JNothing
-        }
-      })
+  private[sql] def jsonNode: JsonNode = {
+    val obj = factory.objectNode()
+    obj.put("operatorName", operatorName)
+    obj.put("numRowsTotal", numRowsTotal)
+    obj.put("numRowsUpdated", numRowsUpdated)
+    obj.put("allUpdatesTimeMs", allUpdatesTimeMs)
+    obj.put("numRowsRemoved", numRowsRemoved)
+    obj.put("allRemovalsTimeMs", allRemovalsTimeMs)
+    obj.put("commitTimeMs", commitTimeMs)
+    obj.put("memoryUsedBytes", memoryUsedBytes)
+    obj.put("numRowsDroppedByWatermark", numRowsDroppedByWatermark)
+    obj.put("numShufflePartitions", numShufflePartitions)
+    obj.put("numStateStoreInstances", numStateStoreInstances)
+    if (!customMetrics.isEmpty) {
+      val metrics = factory.objectNode()
+      val keys = customMetrics.keySet.asScala.toSeq.sorted
+      keys.foreach(k => metrics.put(k, customMetrics.get(k).toLong))
+      obj.set[JsonNode]("customMetrics", metrics)
+    }
+    obj
   }
 
   override def toString: String = prettyJson
@@ -169,30 +167,59 @@ class StreamingQueryProgress private[spark] (
   def processedRowsPerSecond: Double = sources.map(_.processedRowsPerSecond).sum
 
   /** The compact JSON representation of this progress. */
-  def json: String = compact(render(jsonValue))
+  def json: String = JacksonUtils.writeValueAsString(jsonNode)
 
   /** The pretty (i.e. indented) JSON representation of this progress. */
-  def prettyJson: String = pretty(render(jsonValue))
+  def prettyJson: String = JacksonUtils.writeValuePrettyAsString(jsonNode)
 
   override def toString: String = prettyJson
 
-  private[sql] def jsonValue: JValue = {
-    ("id" -> JString(id.toString)) ~
-      ("runId" -> JString(runId.toString)) ~
-      ("name" -> JString(name)) ~
-      ("timestamp" -> JString(timestamp)) ~
-      ("batchId" -> JInt(batchId)) ~
-      ("numInputRows" -> JInt(numInputRows)) ~
-      ("inputRowsPerSecond" -> safeDoubleToJValue(inputRowsPerSecond)) ~
-      ("processedRowsPerSecond" -> safeDoubleToJValue(processedRowsPerSecond)) ~
-      ("durationMs" -> safeMapToJValue[JLong](durationMs, v => JInt(v.toLong))) ~
-      ("eventTime" -> safeMapToJValue[String](eventTime, s => JString(s))) ~
-      ("stateOperators" -> JArray(stateOperators.map(_.jsonValue).toList)) ~
-      ("sources" -> JArray(sources.map(_.jsonValue).toList)) ~
-      ("sink" -> sink.jsonValue) ~
-      ("observedMetrics" -> safeMapToJValue[Row](
-        observedMetrics,
-        row => ToJsonUtil.jsonValue(row)))
+  private[sql] def jsonNode: JsonNode = {
+    val factory = JacksonUtils.defaultNodeFactory
+    val obj = factory.objectNode()
+    obj.put("id", id.toString)
+    obj.put("runId", runId.toString)
+    obj.put("name", name)
+    obj.put("timestamp", timestamp)
+    obj.put("batchId", batchId)
+    obj.put("numInputRows", numInputRows)
+    val inputRowsPerSecondNode = safeDoubleToJsonNode(inputRowsPerSecond)
+    if (!inputRowsPerSecondNode.isMissingNode) {
+      obj.set[JsonNode]("inputRowsPerSecond", inputRowsPerSecondNode)
+    }
+    val processedRowsPerSecondNode = safeDoubleToJsonNode(processedRowsPerSecond)
+    if (!processedRowsPerSecondNode.isMissingNode) {
+      obj.set[JsonNode]("processedRowsPerSecond", processedRowsPerSecondNode)
+    }
+
+    val durationMsNode = safeMapToJsonNode[JLong](durationMs, v => factory.numberNode(v.toLong))
+    if (!durationMsNode.isMissingNode) {
+      obj.set[JsonNode]("durationMs", durationMsNode)
+    }
+
+    val eventTimeNode = safeMapToJsonNode[String](eventTime, v => factory.textNode(v))
+    if (!eventTimeNode.isMissingNode) {
+      obj.set[JsonNode]("eventTime", eventTimeNode)
+    }
+
+
+    val stateOperatorsList = stateOperators.map(_.jsonNode).toList.asJava
+    val stateOperatorsArrayNode = factory.arrayNode(stateOperatorsList.size())
+    stateOperatorsArrayNode.addAll(stateOperatorsList)
+    obj.set[JsonNode]("stateOperators", stateOperatorsArrayNode)
+
+    val sourcesList = sources.map(_.jsonNode).toList.asJava
+    val sourcesArrayNode = factory.arrayNode(sourcesList.size)
+    sourcesArrayNode.addAll(sourcesList)
+    obj.set[JsonNode]("sources", sourcesArrayNode)
+
+    obj.set[JsonNode]("sink", sink.jsonNode)
+    val observedMetricsNode =
+      safeMapToJsonNode[Row](observedMetrics, row => ToJsonUtil.jsonNode(row))
+    if (!observedMetricsNode.isMissingNode) {
+      obj.set[JsonNode]("observedMetrics", observedMetricsNode)
+    }
+    obj
   }
 }
 
@@ -244,28 +271,40 @@ class SourceProgress protected[spark] (
     extends Serializable {
 
   /** The compact JSON representation of this progress. */
-  def json: String = compact(render(jsonValue))
+  def json: String = JacksonUtils.writeValueAsString(jsonNode)
 
   /** The pretty (i.e. indented) JSON representation of this progress. */
-  def prettyJson: String = pretty(render(jsonValue))
+  def prettyJson: String = JacksonUtils.writeValuePrettyAsString(jsonNode)
 
   override def toString: String = prettyJson
 
-  private[sql] def jsonValue: JValue = {
-    ("description" -> JString(description)) ~
-      ("startOffset" -> tryParse(startOffset)) ~
-      ("endOffset" -> tryParse(endOffset)) ~
-      ("latestOffset" -> tryParse(latestOffset)) ~
-      ("numInputRows" -> JInt(numInputRows)) ~
-      ("inputRowsPerSecond" -> safeDoubleToJValue(inputRowsPerSecond)) ~
-      ("processedRowsPerSecond" -> safeDoubleToJValue(processedRowsPerSecond)) ~
-      ("metrics" -> safeMapToJValue[String](metrics, s => JString(s)))
+  private[sql] def jsonNode: JsonNode = {
+    val factory = JacksonUtils.defaultNodeFactory
+    val node = JacksonUtils.createObjectNode
+    node.put("description", description)
+    node.set[JsonNode]("startOffset", tryParseNode(startOffset))
+    node.set[JsonNode]("endOffset", tryParseNode(endOffset))
+    node.set[JsonNode]("latestOffset", tryParseNode(latestOffset))
+    node.put("numInputRows", numInputRows)
+    val inputRowsPerSecondNode = safeDoubleToJsonNode(inputRowsPerSecond)
+    if (!inputRowsPerSecondNode.isMissingNode) {
+      node.set[JsonNode]("inputRowsPerSecond", inputRowsPerSecondNode)
+    }
+    val processedRowsPerSecondNode = safeDoubleToJsonNode(processedRowsPerSecond)
+    if (!processedRowsPerSecondNode.isMissingNode) {
+      node.set[JsonNode]("processedRowsPerSecond", processedRowsPerSecondNode)
+    }
+    val metricsNode = safeMapToJsonNode[String](metrics, s => factory.textNode(s))
+    if (!metricsNode.isMissingNode) {
+      node.set[JsonNode]("metrics", metricsNode)
+    }
+    node
   }
 
-  private def tryParse(json: String) = try {
-    parse(json)
+  private def tryParseNode(json: String): JsonNode = try {
+    JacksonUtils.readTree(json)
   } catch {
-    case NonFatal(e) => JString(json)
+    case NonFatal(_) => JacksonUtils.defaultNodeFactory.textNode(json)
   }
 }
 
@@ -293,17 +332,23 @@ class SinkProgress protected[spark] (
   }
 
   /** The compact JSON representation of this progress. */
-  def json: String = compact(render(jsonValue))
+  def json: String = JacksonUtils.writeValueAsString(jsonNode)
 
   /** The pretty (i.e. indented) JSON representation of this progress. */
-  def prettyJson: String = pretty(render(jsonValue))
+  def prettyJson: String = JacksonUtils.writeValuePrettyAsString(jsonNode)
 
   override def toString: String = prettyJson
 
-  private[sql] def jsonValue: JValue = {
-    ("description" -> JString(description)) ~
-      ("numOutputRows" -> JInt(numOutputRows)) ~
-      ("metrics" -> safeMapToJValue[String](metrics, s => JString(s)))
+  private[sql] def jsonNode: JsonNode = {
+    val factory = JacksonUtils.defaultNodeFactory
+    val obj = factory.objectNode()
+    obj.put("description", description)
+    obj.put("numOutputRows", numOutputRows)
+    val metricsNode = safeMapToJsonNode[String](metrics, s => factory.textNode(s))
+    if (!metricsNode.isMissingNode) {
+      obj.set[JsonNode]("metrics", metricsNode)
+    }
+    obj
   }
 }
 
@@ -318,14 +363,22 @@ private[sql] object SinkProgress {
 }
 
 private object SafeJsonSerializer {
-  def safeDoubleToJValue(value: Double): JValue = {
-    if (value.isNaN || value.isInfinity) JNothing else JDouble(value)
+  def safeDoubleToJsonNode(value: Double): JsonNode = {
+    val factory = JacksonUtils.defaultNodeFactory
+    if (value.isNaN || value.isInfinity) {
+      factory.missingNode()
+    } else {
+      factory.numberNode(value)
+    }
   }
 
   /** Convert map to JValue while handling empty maps. Also, this sorts the keys. */
-  def safeMapToJValue[T](map: ju.Map[String, T], valueToJValue: T => JValue): JValue = {
-    if (map.isEmpty) return JNothing
+  def safeMapToJsonNode[T](map: ju.Map[String, T], valueToJsonNode: T => JsonNode): JsonNode = {
+    val factory = JacksonUtils.defaultNodeFactory
+    if (map.isEmpty) return factory.missingNode()
     val keys = map.asScala.keySet.toSeq.sorted
-    keys.map { k => k -> valueToJValue(map.get(k)): JObject }.reduce(_ ~ _)
+    val node = factory.objectNode()
+    keys.foreach { k => node.set[JsonNode](k, valueToJsonNode(map.get(k))) }
+    node
   }
 }

connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala

@@ -34,8 +34,6 @@ import io.grpc.protobuf.services.ProtoReflectionService
 import io.grpc.stub.StreamObserver
 import org.apache.commons.lang3.StringUtils
 import org.apache.commons.lang3.exception.ExceptionUtils
-import org.json4s.JsonDSL._
-import org.json4s.jackson.JsonMethods.{compact, render}
 
 import org.apache.spark.{SparkEnv, SparkException, SparkThrowable}
 import org.apache.spark.api.python.PythonException
@@ -45,6 +43,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.connect.config.Connect.{CONNECT_GRPC_BINDING_PORT, CONNECT_GRPC_MAX_INBOUND_MESSAGE_SIZE, CONNECT_JVM_STACK_TRACE_MAX_SIZE}
 import org.apache.spark.sql.internal.SQLConf.PYSPARK_JVM_STACKTRACE_ENABLED
+import org.apache.spark.util.JacksonUtils
 
 /**
  * The SparkConnectService implementation.
@@ -80,7 +79,9 @@ class SparkConnectService(debug: Boolean)
       .newBuilder()
       .setReason(st.getClass.getName)
       .setDomain("org.apache.spark")
-      .putMetadata("classes", compact(render(allClasses(st.getClass).map(_.getName))))
+      .putMetadata(
+        "classes",
+        JacksonUtils.writeValueAsString(allClasses(st.getClass).map(_.getName)))
 
     lazy val stackTrace = Option(ExceptionUtils.getStackTrace(st))
     val withStackTrace = if (stackTraceEnabled && stackTrace.nonEmpty) {