Spark-19726: alwaysNullable as a parameter for getSchema

Author: YIHAODIAN\wangshuangshuang

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

@@ -61,7 +61,7 @@ object JDBCRDD extends Logging {
       try {
         val rs = statement.executeQuery()
         try {
-          JdbcUtils.getSchema(rs, dialect)
+          JdbcUtils.getSchema(rs, dialect, true)
         } finally {
           rs.close()
         }

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

@@ -264,19 +264,19 @@ object JdbcUtils extends Logging {
   }
 
   /**
-   * Takes a [[ResultSet]] and returns its Catalyst schema.
-   *
-   * @return A [[StructType]] giving the Catalyst schema.
-   * @throws SQLException if the schema contains an unsupported type.
-   */
-  def getSchema(resultSet: ResultSet, dialect: JdbcDialect): StructType = {
+    * Takes a [[ResultSet]] and returns its Catalyst schema.
+    *
+    * @param alwaysNullable - If true, spark will propagate null
+    *                       to underlying DB engine instead of using type default value
+    * @return A [[StructType]] giving the Catalyst schema.
+    * @throws SQLException if the schema contains an unsupported type.
+    */
+  def getSchema(resultSet: ResultSet,
+                dialect: JdbcDialect,
+                alwaysNullable: Boolean = false): StructType = {
     val rsmd = resultSet.getMetaData
     val ncols = rsmd.getColumnCount
     val fields = new Array[StructField](ncols)
-    // if true, spark will propagate null to underlying
-    // DB engine instead of using type default value
-    val alwaysNullable = true
-
     var i = 0
     while (i < ncols) {
       val columnName = rsmd.getColumnLabel(i + 1)
@@ -294,12 +294,10 @@ object JdbcUtils extends Logging {
             rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
         }
       }
-      val nullable = {
-        if (alwaysNullable)
-          alwaysNullable
-        else
-          rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
-      }
+      val nullable = if (alwaysNullable)
+        alwaysNullable
+      else
+        rsmd.isNullable(i + 1) != ResultSetMetaData.columnNoNulls
       val metadata = new MetadataBuilder()
         .putString("name", columnName)
         .putLong("scale", fieldScale)

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

@@ -20,10 +20,11 @@ package org.apache.spark.sql.jdbc
 import java.sql.{Date, DriverManager, Timestamp}
 import java.util.Properties
 
-import org.apache.spark.SparkException
-
 import scala.collection.JavaConverters.propertiesAsScalaMapConverter
+
 import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.SparkException
 import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode}
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}