[SPARK-55716][SQL][FOLLOWUP] Simplify NOT NULL preservation by skipping asNullable in resolveRelation

Instead of calling `dataSchema.asNullable` in `resolveRelation()` and then restoring nullability from the user schema in `CreateDataSourceTableCommand`, add a `forceNullable` parameter to `resolveRelation()` and pass `false` from the create-table path.

This eliminates the `restoreNullability` and `restoreDataTypeNullability` helper methods, simplifying the code significantly (-37 lines net).

The read path is unaffected: `resolveRelation()` defaults to `forceNullable = true` (preserving SPARK-13738 safety). Only the create-table path skips `asNullable` so the catalog stores the user's original nullability constraints.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: yaooqinn

sql/core/src/main/scala/org/apache/spark/sql/execution/command/createDataSourceTables.scala

@@ -28,8 +28,9 @@ import org.apache.spark.sql.classic.ClassicConversions.castToImpl
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.CommandExecutionMode
 import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources.BaseRelation
-import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
+import org.apache.spark.sql.types.StructType
 import org.apache.spark.util.ArrayImplicits._
 
 /**
@@ -79,7 +80,9 @@ case class CreateDataSourceTableCommand(table: CatalogTable, ignoreIfExists: Boo
         bucketSpec = table.bucketSpec,
         options = table.storage.properties ++ pathOption,
         // As discussed in SPARK-19583, we don't check if the location is existed
-        catalogTable = Some(tableWithDefaultOptions)).resolveRelation(checkFilesExist = false)
+        catalogTable = Some(tableWithDefaultOptions))
+        .resolveRelation(checkFilesExist = false,
+          forceNullable = !sessionState.conf.getConf(SQLConf.FILE_SOURCE_INSERT_ENFORCE_NOT_NULL))
 
     val partitionColumnNames = if (table.schema.nonEmpty) {
       table.partitionColumnNames
@@ -107,17 +110,8 @@ case class CreateDataSourceTableCommand(table: CatalogTable, ignoreIfExists: Boo
         table.copy(schema = new StructType(), partitionColumnNames = Nil)
 
       case _ =>
-        // Merge nullability from the user-specified schema into the resolved schema.
-        // DataSource.resolveRelation() calls dataSchema.asNullable which strips NOT NULL
-        // constraints. We restore nullability from the original user schema while keeping
-        // the resolved data types (which may include CharVarchar normalization, metadata, etc.)
-        val resolvedSchema = if (table.schema.nonEmpty) {
-          restoreNullability(dataSource.schema, table.schema)
-        } else {
-          dataSource.schema
-        }
         table.copy(
-          schema = resolvedSchema,
+          schema = dataSource.schema,
           partitionColumnNames = partitionColumnNames,
           // If metastore partition management for file source tables is enabled, we start off with
           // partition provider hive, but no partitions in the metastore. The user has to call
@@ -132,38 +126,6 @@ case class CreateDataSourceTableCommand(table: CatalogTable, ignoreIfExists: Boo
     Seq.empty[Row]
   }
 
-  /**
-   * Recursively restores nullability from the original user-specified schema into
-   * the resolved schema. The resolved schema's data types are preserved (they may
-   * contain CharVarchar normalization, metadata, etc.), but nullability flags
-   * (top-level and nested) are taken from the original schema.
-   */
-  private def restoreNullability(resolved: StructType, original: StructType): StructType = {
-    val originalFields = original.fields.map(f => f.name -> f).toMap
-    StructType(resolved.fields.map { resolvedField =>
-      originalFields.get(resolvedField.name) match {
-        case Some(origField) =>
-          resolvedField.copy(
-            nullable = origField.nullable,
-            dataType = restoreDataTypeNullability(resolvedField.dataType, origField.dataType))
-        case None => resolvedField
-      }
-    })
-  }
-
-  private def restoreDataTypeNullability(resolved: DataType, original: DataType): DataType = {
-    (resolved, original) match {
-      case (r: StructType, o: StructType) => restoreNullability(r, o)
-      case (ArrayType(rElem, _), ArrayType(oElem, oNull)) =>
-        ArrayType(restoreDataTypeNullability(rElem, oElem), oNull)
-      case (MapType(rKey, rVal, _), MapType(oKey, oVal, oValNull)) =>
-        MapType(
-          restoreDataTypeNullability(rKey, oKey),
-          restoreDataTypeNullability(rVal, oVal),
-          oValNull)
-      case _ => resolved
-    }
-  }
 }
 
 /**

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

@@ -362,7 +362,10 @@ case class DataSource(
    *                        is considered as a non-streaming file based data source. Since we know
    *                        that files already exist, we don't need to check them again.
    */
-  def resolveRelation(checkFilesExist: Boolean = true, readOnly: Boolean = false): BaseRelation = {
+  def resolveRelation(
+      checkFilesExist: Boolean = true,
+      readOnly: Boolean = false,
+      forceNullable: Boolean = true): BaseRelation = {
     val relation = (providingInstance(), userSpecifiedSchema) match {
       // TODO: Throw when too much is given.
       case (dataSource: SchemaRelationProvider, Some(schema)) =>
@@ -436,7 +439,7 @@ case class DataSource(
         HadoopFsRelation(
           fileCatalog,
           partitionSchema = partitionSchema,
-          dataSchema = dataSchema.asNullable,
+          dataSchema = if (forceNullable) dataSchema.asNullable else dataSchema,
           bucketSpec = bucketSpec,
           format,
           caseInsensitiveOptions)(sparkSession)