[postgres]Improve pattern matching for prepared

Author: parisni

spark-postgres/src/main/scala/io/frama/parisni/spark/postgres/PGTool.scala

@@ -392,17 +392,14 @@ object PGTool extends java.io.Serializable with LazyLogging {
 
   def parametrize(st: PreparedStatement, params: List[Any]) = {
     for ((obj, i) <- params.zipWithIndex) {
-      obj.getClass.getCanonicalName match {
-        case "java.lang.String" => st.setString(i + 1, obj.asInstanceOf[String])
-        case "java.lang.Boolean" =>
-          st.setBoolean(i + 1, obj.asInstanceOf[Boolean])
-        case "java.lang.Long"    => st.setLong(i + 1, obj.asInstanceOf[Long])
-        case "java.lang.Integer" => st.setInt(i + 1, obj.asInstanceOf[Int])
-        case "java.math.BigDecimal" =>
-          st.setDouble(i + 1, obj.asInstanceOf[Double])
-        case "java.sql.Date" => st.setDate(i + 1, obj.asInstanceOf[Date])
-        case "java.sql.Timestamp" =>
-          st.setTimestamp(i + 1, obj.asInstanceOf[Timestamp])
+      obj match {
+        case s: String               => st.setString(i + 1, s)
+        case b: Boolean              => st.setBoolean(i + 1, b)
+        case l: Long                 => st.setLong(i + 1, l)
+        case i: Integer              => st.setInt(i + 1, i)
+        case b: java.math.BigDecimal => st.setDouble(i + 1, b.doubleValue())
+        case d: java.sql.Date        => st.setDate(i + 1, d)
+        case t: Timestamp            => st.setTimestamp(i + 1, t)
         case _ =>
           throw new UnsupportedEncodingException(
             obj.getClass.getCanonicalName + " type not yet supported for prepared statements")

spark-postgres/src/test/scala/io/frama/parisni/spark/postgres/CrudTest.scala

@@ -0,0 +1,47 @@
+package io.frama.parisni.spark.postgres
+
+import org.apache.spark.sql.{DataFrame, QueryTest}
+import org.junit.Test
+
+class CrudTest extends QueryTest with SparkSessionTestWrapper {
+
+  @Test
+  def verifyParametrised(): Unit = {
+    import spark.implicits._
+
+    val df: DataFrame = (("bob", 2, true) ::
+      Nil).toDF("colstring", "colint", "colboolean")
+    getPgTool().tableCreate("test_crud", df.schema)
+    val query =
+      "insert into test_crud (colstring, colint, colboolean) values (?, ? ,?)"
+    getPgTool().sqlExec(query, List("bob", 1, false))
+    val output = spark.read
+      .format("io.frama.parisni.spark.postgres")
+      .option("host", "localhost")
+      .option("port", pg.getEmbeddedPostgres.getPort)
+      .option("database", "postgres")
+      .option("user", "postgres")
+      .option("query", "select * from test_crud")
+      .load
+    val wanted =
+      (("bob", 1, false) :: Nil).toDF("colstring", "colint", "colboolean")
+    checkAnswer(output, wanted)
+
+  }
+
+  @Test
+  def verifyParametrisedWithResult(): Unit = {
+    import spark.implicits._
+
+    val df: DataFrame = (("bob", 2, true) ::
+      Nil).toDF("colstring", "colint", "colboolean")
+    getPgTool().tableCreate("test_crud2", df.schema)
+    val query =
+      "insert into test_crud2 (colstring, colint, colboolean) values (?, ? ,?) returning colstring, colint, colboolean"
+    val result = getPgTool().sqlExecWithResult(query, List("bob", 1, false))
+    val wanted =
+      (("bob", 1, false) :: Nil).toDF("colstring", "colint", "colboolean")
+    checkAnswer(result, wanted)
+  }
+
+}