diff --git a/pom.xml b/pom.xml
index 1e3e8367..eb40208b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -46,7 +46,6 @@
         <scalaPluginVersion>4.3.0</scalaPluginVersion>
         <jackson.version>2.13.2</jackson.version>
         <jackson.databind.version>2.13.4.2</jackson.databind.version>
-        <jackson.module.scala.version>2.13.5</jackson.module.scala.version>
 
     </properties>
     <dependencyManagement>
@@ -133,11 +132,6 @@
             <artifactId>jackson-annotations</artifactId>
             <version>${jackson.version}</version>
         </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.module</groupId>
-            <artifactId>jackson-module-scala_2.12</artifactId>
-            <version>${jackson.module.scala.version}</version>
-        </dependency>
 
         <!-- Test -->
         <dependency>
diff --git a/src/main/scala/com/snowflake/snowpark/Session.scala b/src/main/scala/com/snowflake/snowpark/Session.scala
index 90f7b83e..1beac1c0 100644
--- a/src/main/scala/com/snowflake/snowpark/Session.scala
+++ b/src/main/scala/com/snowflake/snowpark/Session.scala
@@ -1,8 +1,5 @@
 package com.snowflake.snowpark
 
-import com.fasterxml.jackson.databind.json.JsonMapper
-import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
-
 import java.io.{File, FileInputStream, FileNotFoundException}
 import java.net.URI
 import java.sql.{Connection, Date, Time, Timestamp}
@@ -29,7 +26,6 @@ import net.snowflake.client.jdbc.{SnowflakeConnectionV1, SnowflakeDriver, Snowfl
 import scala.concurrent.{ExecutionContext, Future}
 import scala.collection.JavaConverters._
 import scala.reflect.runtime.universe.TypeTag
-import scala.util.Try
 
 /**
  *
@@ -65,11 +61,6 @@ import scala.util.Try
  * @since 0.1.0
  */
 class Session private (private[snowpark] val conn: ServerConnection) extends Logging {
-  private val jsonMapper = JsonMapper
-    .builder()
-    .addModule(DefaultScalaModule)
-    .build() :: ClassTagExtensions
-
   private val STAGE_PREFIX = "@"
   // URI and file name with md5
   private val classpathURIs = new ConcurrentHashMap[URI, Option[String]]().asScala
@@ -397,7 +388,7 @@ class Session private (private[snowpark] val conn: ServerConnection) extends Log
    *         successful, or `None` otherwise.
    */
   private def parseJsonString(jsonString: String): Option[Map[String, Any]] = {
-    Try(jsonMapper.readValue[Map[String, Any]](jsonString)).toOption
+    Utils.jsonToMap(jsonString)
   }
 
   /**
@@ -408,7 +399,7 @@ class Session private (private[snowpark] val conn: ServerConnection) extends Log
    *         or `None` otherwise.
    */
   private def toJsonString(map: Map[String, Any]): Option[String] = {
-    Try(jsonMapper.writeValueAsString(map)).toOption
+    Utils.mapToJson(map)
   }
 
   /*
diff --git a/src/main/scala/com/snowflake/snowpark/internal/Utils.scala b/src/main/scala/com/snowflake/snowpark/internal/Utils.scala
index d464566c..f84459ee 100644
--- a/src/main/scala/com/snowflake/snowpark/internal/Utils.scala
+++ b/src/main/scala/com/snowflake/snowpark/internal/Utils.scala
@@ -1,12 +1,7 @@
 package com.snowflake.snowpark.internal
 
 import com.snowflake.snowpark.Column
-import com.snowflake.snowpark.internal.analyzer.{
-  Attribute,
-  LogicalPlan,
-  TableFunctionExpression,
-  singleQuote
-}
+import com.snowflake.snowpark.internal.analyzer.{Attribute, LogicalPlan, TableFunctionExpression, singleQuote}
 
 import java.io.{File, FileInputStream}
 import java.lang.invoke.SerializedLambda
@@ -14,7 +9,10 @@ import java.security.{DigestInputStream, MessageDigest}
 import java.util.Locale
 import com.snowflake.snowpark.udtf.UDTF
 import net.snowflake.client.jdbc.SnowflakeSQLException
+import net.snowflake.client.jdbc.internal.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import net.snowflake.client.jdbc.internal.fasterxml.jackson.databind.node.JsonNodeType
 
+import scala.collection.JavaConverters.asScalaIteratorConverter
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.util.Random
@@ -447,4 +445,77 @@ object Utils extends Logging {
       case _ => throw ErrorMessage.DF_JOIN_WITH_WRONG_ARGUMENT()
     }
   }
+
+  private val objectMapper = new ObjectMapper()
+
+  private[snowpark] def jsonToMap(jsonString: String): Option[Map[String, Any]] = {
+    try {
+      val node = objectMapper.readTree(jsonString)
+      assert(node.getNodeType == JsonNodeType.OBJECT)
+      Some(jsonToScala(node).asInstanceOf[Map[String, Any]])
+    } catch {
+      case ex: Exception =>
+        logError(ex.getMessage)
+        None
+    }
+  }
+
+  private def jsonToScala(node: JsonNode): Any = {
+    node.getNodeType match {
+      case JsonNodeType.STRING => node.asText()
+      case JsonNodeType.NULL => null
+      case JsonNodeType.OBJECT =>
+        node.fields().asScala.map(entry => {
+          entry.getKey -> jsonToScala(entry.getValue)
+        }).toMap
+      case JsonNodeType.ARRAY =>
+        node.elements().asScala.map(entry => jsonToScala(entry)).toList
+      case JsonNodeType.BOOLEAN => node.asBoolean()
+      case JsonNodeType.NUMBER => node.numberValue()
+      case other =>
+        throw new UnsupportedOperationException(s"Unsupported Type: ${other.name()}")
+    }
+  }
+
+  private[snowpark] def mapToJson(map: Map[String, Any]): Option[String] = {
+    try {
+      Some(scalaToJson(map))
+    } catch {
+      case ex: Exception =>
+        logError(ex.getMessage)
+        None
+    }
+  }
+
+  private def scalaToJson(node: Any): String = {
+    node match {
+      case n: Map[String, Any] =>
+        val mapRes = n.map {
+          case (key, value: Map[String, Any]) =>
+            s""""${key}": ${scalaToJson(value)}"""
+          case (key, value: List[Any]) =>
+            s""""${key}": ${scalaToJson(value)}"""
+          case (key, value: Number) =>
+            s""""${key}": ${value}"""
+          case (key, value: String) =>
+            s""""${key}": "${value}""""
+          case (key, value: Boolean) =>
+            s""""${key}": ${value}"""
+          case (key, None) =>
+            s""""${key}": null"""
+          case other =>
+            throw new UnsupportedOperationException(s"Unsupported Type: ${other.getClass}")
+        }
+        s"{${mapRes.mkString(", ")}}"
+      case n: List[Any] =>
+        val listRes = n.map {
+          case v: List[Any] => scalaToJson(v)
+          case v: Map[String, Any] => scalaToJson(v)
+          case v => v.toString
+        }
+        s"[${listRes.mkString(", ")}]"
+      case other =>
+        throw new UnsupportedOperationException(s"Unsupported Type: ${other.getClass}")
+    }
+  }
 }
diff --git a/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala b/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala
index 2067212f..d02984da 100644
--- a/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala
+++ b/src/test/scala/com/snowflake/snowpark/UtilsSuite.scala
@@ -672,6 +672,28 @@ class UtilsSuite extends SNTestBase {
     assert(Utils.quoteForOption("FALSE").equals("FALSE"))
     assert(Utils.quoteForOption("abc").equals("'abc'"))
   }
+
+  test("Scala and Json format transformation") {
+    val map = Map(
+      "integerKey" -> 1.1,
+      "stringKey" -> "stringValue",
+      "nestedMap" -> Map(
+        "insideKey" -> "stringValue",
+        "insideList" -> Seq(1, 2, 3)
+      ),
+      "nestedList" -> Seq(
+        1,
+        Map(
+          "nestedKey" -> "nestedValue"
+        ),
+        List(1, 2, 3)
+      )
+    )
+    val jsonString = Utils.mapToJson(map)
+    val readMap = Utils.jsonToMap(jsonString.getOrElse(""))
+    val transformedString = Utils.mapToJson(readMap.getOrElse(Map()))
+    assert(jsonString.equals(transformedString))
+  }
 }
 
 object LoggingTester extends Logging {