Spark-19726: add a new boolean parameter alwaysNullable to getSchema

Author: YIHAODIAN\wangshuangshuang

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

@@ -61,13 +61,7 @@ object JDBCRDD extends Logging {
       try {
         val rs = statement.executeQuery()
         try {
-          val metaStructType = JdbcUtils.getSchema(rs, dialect)
-          StructType(metaStructType.map(f =>
-            if(f.nullable)
-              f
-            else
-              StructField(f.name, f.dataType, true, f.metadata)
-          ))
+          JdbcUtils.getSchema(rs, dialect)
         } finally {
           rs.close()
         }

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

@@ -273,6 +273,9 @@ object JdbcUtils extends Logging {
     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)
@@ -290,7 +293,7 @@ object JdbcUtils extends Logging {
             rsmd.getClass.getName == "org.apache.hive.jdbc.HiveResultSetMetaData" => true
         }
       }
-      val nullable = 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,10 @@ package org.apache.spark.sql.jdbc
 import java.sql.{Date, DriverManager, Timestamp}
 import java.util.Properties
 
-import scala.collection.JavaConverters.propertiesAsScalaMapConverter
+import org.apache.spark.SparkException
 
+import scala.collection.JavaConverters.propertiesAsScalaMapConverter
 import org.scalatest.BeforeAndAfter
-
 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}
@@ -506,4 +506,11 @@ class JDBCWriteSuite extends SharedSQLContext with BeforeAndAfter {
         "schema struct<name:string,id:int>"))
     }
   }
+
+  test("SPARK-19726: INSERT null to a NOT NULL column") {
+    val e = intercept[SparkException] {
+      sql("INSERT INTO PEOPLE1 values (null, null)")
+    }.getMessage
+    assert(e.contains("NULL not allowed for column \"NAME\""))
+  }
 }