[Spark-10625] [SQL] Spark SQL JDBC read/write is unable to handle JDBC Drivers that adds unserializable objects into connection properties

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala

@@ -38,6 +38,9 @@ private[sql] case class JDBCPartitioningInfo(
     numPartitions: Int)
 
 private[sql] object JDBCRelation extends Logging {
+
+  import scala.collection.JavaConverters._
+
   /**
    * Given a partitioning schematic (a column of integral type, a number of
    * partitions, and upper and lower bounds on the column's value), generate
@@ -99,6 +102,16 @@ private[sql] object JDBCRelation extends Logging {
     }
     ans.toArray
   }
+
+  def getEffectiveProperties(
+      connectionProperties: Properties,
+      extraOptions: scala.collection.Map[String, String] = Map()): Properties = {
+    val props = new Properties()
+    props.putAll(extraOptions.asJava)
+    // connectionProperties should override settings in extraOptions
+    props.putAll(connectionProperties)
+    props
+  }
 }
 
 private[sql] case class JDBCRelation(
@@ -127,7 +140,7 @@ private[sql] case class JDBCRelation(
       sparkSession.sparkContext,
       schema,
       url,
-      properties,
+      JDBCRelation.getEffectiveProperties(properties),
       table,
       requiredColumns,
       filters,

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -63,7 +63,7 @@ object JdbcUtils extends Logging {
         throw new IllegalStateException(
           s"Did not find registered driver with class $driverClass")
       }
-      driver.connect(url, properties)
+      driver.connect(url, JDBCRelation.getEffectiveProperties(properties))
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -686,6 +686,13 @@ class JDBCSuite extends SparkFunSuite
       Some(DecimalType(DecimalType.MAX_PRECISION, 10)))
   }
 
+  test("SPARK-10625: JDBC read should allow driver to insert unserializable into properties") {
+    UnserializableDriverHelper.replaceDriverDuring {
+      assert(sqlContext.read.jdbc(
+        urlWithUserAndPass, "TEST.PEOPLE", new Properties).collect().length === 3)
+    }
+  }
+
   test("table exists query by jdbc dialect") {
     val MySQL = JdbcDialects.get("jdbc:mysql://127.0.0.1/db")
     val Postgres = JdbcDialects.get("jdbc:postgresql://127.0.0.1/db")

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala

@@ -196,4 +196,12 @@ class JDBCWriteSuite extends SharedSQLContext with BeforeAndAfter {
     assert(2 === spark.read.jdbc(url1, "TEST.PEOPLE1", properties).count())
     assert(2 === spark.read.jdbc(url1, "TEST.PEOPLE1", properties).collect()(0).length)
   }
+
+  test("SPARK-10625: JDBC write should allow driver to insert unserializable into properties") {
+    UnserializableDriverHelper.replaceDriverDuring {
+      sql("INSERT INTO TABLE PEOPLE1 SELECT * FROM PEOPLE")
+      assert(2 === sqlContext.read.jdbc(url1, "TEST.PEOPLE1", properties).count)
+      assert(2 === sqlContext.read.jdbc(url1, "TEST.PEOPLE1", properties).collect()(0).length)
+    }
+  }
 }

sql/core/src/test/scala/org/apache/spark/sql/jdbc/UnserializableDriverHelper.scala

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.jdbc
+
+import java.sql.{Connection, DriverManager}
+import java.util.Properties
+import java.util.logging.Logger
+
+object UnserializableDriverHelper {
+
+  import scala.collection.JavaConverters._
+
+  def replaceDriverDuring[T](f: => T): T = {
+    object UnserializableH2Driver extends org.h2.Driver {
+
+      override def connect(url: String, info: Properties): Connection = {
+
+        val result = super.connect(url, info)
+        info.put("unserializableDriver", this)
+        result
+      }
+
+      override def getParentLogger: Logger = null
+    }
+
+    val oldDrivers = DriverManager.getDrivers.asScala.toList.filter(_.acceptsURL("jdbc:h2:"))
+    oldDrivers.foreach(DriverManager.deregisterDriver)
+    DriverManager.registerDriver(UnserializableH2Driver)
+
+    val result = try {
+      f
+    } finally {
+      DriverManager.deregisterDriver(UnserializableH2Driver)
+      oldDrivers.foreach(DriverManager.registerDriver)
+    }
+    result
+  }
+}