[SPARK-16472][SQL] Force user specified schema to the nullable one

python/pyspark/sql/streaming.py

@@ -839,7 +839,7 @@ def _test():
     globs['sqlContext'] = SQLContext.getOrCreate(spark.sparkContext)
     globs['sdf'] = \
         spark.readStream.format('text').load('python/test_support/sql/streaming')
-    globs['sdf_schema'] = StructType([StructField("data", StringType(), False)])
+    globs['sdf_schema'] = StructType([StructField("data", StringType(), True)])
     globs['df'] = \
         globs['spark'].readStream.format('text').load('python/test_support/sql/streaming')
 

sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

@@ -60,7 +60,7 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
    * @since 1.4.0
    */
   def schema(schema: StructType): DataFrameReader = {
-    this.userSpecifiedSchema = Option(schema)
+    this.userSpecifiedSchema = Option(schema.asNullable)
     this
   }
 

sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala

@@ -53,7 +53,7 @@ final class DataStreamReader private[sql](sparkSession: SparkSession) extends Lo
    * @since 2.0.0
    */
   def schema(schema: StructType): DataStreamReader = {
-    this.userSpecifiedSchema = Option(schema)
+    this.userSpecifiedSchema = Option(schema.asNullable)
     this
   }
 

sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala

@@ -1191,6 +1191,18 @@ class FileStreamSourceSuite extends FileStreamSourceTest {
       newSource.getBatch(None, FileStreamSourceOffset(1))
     }
   }
+
+  test("Force user specified schema to the nullable") {
+    withTempDir { src =>
+      stringToFile(new File(src, "1"), "a\nb\nc")
+      val userSchema =
+        StructType(StructField("a", StringType, nullable = false) :: Nil)
+      val schema = createFileStreamSourceAndGetSchema(
+        format = Some("parquet"), path = Some(src.getCanonicalPath), schema = Some(userSchema))
+      assert(schema != userSchema)
+      assert(schema === userSchema.asNullable)
+    }
+  }
 }
 
 class FileStreamSourceStressTestSuite extends FileStreamSourceTest {

sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala

@@ -635,4 +635,12 @@ class DataFrameReaderWriterSuite extends QueryTest with SharedSQLContext with Be
       checkAnswer(spark.table("t"), Row(1, "a") :: Row(2, "b") :: Nil)
     }
   }
+
+  test("SPARK-16472: Force user specified schema to the nullable") {
+    val schema = new StructType().add("s", StringType, nullable = false)
+    val nullableSchema = schema.asNullable
+    val df = spark.read.schema(schema).text()
+    assert(df.schema != schema)
+    assert(df.schema == nullableSchema)
+  }
 }