[SPARK-28651][SS]Force the schema of Streaming file source to be nullable

python/pyspark/sql/streaming.py

@@ -1155,7 +1155,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/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -1140,6 +1140,15 @@ object SQLConf {
       .timeConf(TimeUnit.MILLISECONDS)
       .createWithDefault(TimeUnit.MINUTES.toMillis(10)) // 10 minutes
 
+  val FILE_SOURCE_SCHEMA_FORCE_NULLABLE =
+    buildConf("spark.sql.streaming.fileSource.schema.forceNullable")
+      .internal()
+      .doc("When true, force the schema of streaming file source to be nullable (including all " +
+        "the fields). Otherwise, the schema might not be compatible with actual data, which " +
+        "leads to corruptions.")
+      .booleanConf
+      .createWithDefault(true)
+
   val STREAMING_SCHEMA_INFERENCE =
     buildConf("spark.sql.streaming.schemaInference")
       .internal()

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

@@ -254,9 +254,12 @@ case class DataSource(
             checkAndGlobPathIfNecessary(checkEmptyGlobPath = false, checkFilesExist = false)
           createInMemoryFileIndex(globbedPaths)
         })
+        val forceNullable =
+          sparkSession.sessionState.conf.getConf(SQLConf.FILE_SOURCE_SCHEMA_FORCE_NULLABLE)
+        val sourceDataSchema = if (forceNullable) dataSchema.asNullable else dataSchema
         SourceInfo(
           s"FileSource[$path]",
-          StructType(dataSchema ++ partitionSchema),
+          StructType(sourceDataSchema ++ partitionSchema),
           partitionSchema.fieldNames)
 
       case _ =>

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

@@ -1577,6 +1577,25 @@ class FileStreamSourceSuite extends FileStreamSourceTest {
       )
     }
   }
+
+  test("SPARK-28651: force streaming file source to be nullable") {
+    withTempDir { temp =>
+      val schema = StructType(Seq(StructField("foo", LongType, false)))
+      val nullableSchema = StructType(Seq(StructField("foo", LongType, true)))
+      val streamingSchema = spark.readStream.schema(schema).json(temp.getCanonicalPath).schema
+      assert(nullableSchema === streamingSchema)
+
+      // Verify we have the same behavior as batch DataFrame.
+      val batchSchema = spark.read.schema(schema).json(temp.getCanonicalPath).schema
+      assert(batchSchema === streamingSchema)
+
+      // Verify the flag works
+      withSQLConf(SQLConf.FILE_SOURCE_SCHEMA_FORCE_NULLABLE.key -> "false") {
+        val streamingSchema = spark.readStream.schema(schema).json(temp.getCanonicalPath).schema
+        assert(schema === streamingSchema)
+      }
+    }
+  }
 }
 
 class FileStreamSourceStressTestSuite extends FileStreamSourceTest {