add nested check

Author: ulysses-you

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

@@ -29,7 +29,7 @@ import org.apache.spark.sql.execution.command.{CreateDataSourceTableCommand, Cre
 import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources.InsertableRelation
-import org.apache.spark.sql.types.{AtomicType, DataTypes, StructType}
+import org.apache.spark.sql.types._
 import org.apache.spark.sql.util.SchemaUtils
 
 /**
@@ -394,38 +394,55 @@ case class PreprocessTableInsertion(conf: SQLConf) extends Rule[LogicalPlan] {
 }
 
 /**
- * SPARK-28443: Spark sql add exception when create field type NullType
+ * SPARK-28443: fail the DDL command if it creates a table with null-type columns
  */
 object DDLCheck extends (LogicalPlan => Unit) {
 
   def failAnalysis(msg: String): Unit = { throw new AnalysisException(msg) }
 
-  def throwWhenExistsNullType(schema: StructType): Unit = {
-    if (schema.exists(f => DataTypes.NullType.sameType(f.dataType))) {
-      failAnalysis("DataType NullType is not supported for create table ")
+  def throwWhenExistsNullType(dataType: DataType): Unit = {
+    dataType match {
+      case ArrayType(elementType, _) =>
+        throwWhenExistsNullType(elementType)
+
+      case MapType(keyType, valueType, _) =>
+        throwWhenExistsNullType(keyType)
+        throwWhenExistsNullType(valueType)
+
+      case StructType(fields) =>
+        fields.foreach{field => throwWhenExistsNullType(field.dataType)}
+
+      case other if other == NullType =>
+        failAnalysis("DataType NullType is not supported for create table")
+
+      case _ => // OK
     }
   }
 
+  def checkSchema(schema: StructType): Unit = {
+    schema.foreach{field => throwWhenExistsNullType(field.dataType)}
+  }
+
   override def apply(plan: LogicalPlan): Unit = {
     plan.foreach {
       case CreateTable(tableDesc, _, _) =>
-        throwWhenExistsNullType(tableDesc.schema)
+        checkSchema(tableDesc.schema)
 
       case CreateV2Table(_, _, tableSchema, _, _, _) =>
-        throwWhenExistsNullType(tableSchema)
+        checkSchema(tableSchema)
 
       // DataSourceAnalysis will convert CreateTable to CreateDataSourceTableCommand before check
       case CreateDataSourceTableCommand(table, _) =>
-        throwWhenExistsNullType(table.schema)
+        checkSchema(table.schema)
 
       // HiveAnalysis will convert CreateTable to CreateTableCommand before check
       case CreateTableCommand(table, _) =>
-        throwWhenExistsNullType(table.schema)
+        checkSchema(table.schema)
 
       case ReplaceTable(_, _, tableSchema, _, _, _) =>
-        throwWhenExistsNullType(tableSchema)
+        checkSchema(tableSchema)
 
-      case _ => // OK
+      case _ => // skip
     }
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/internal/CatalogSuite.scala

@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionInfo}
 import org.apache.spark.sql.catalyst.plans.logical.Range
 import org.apache.spark.sql.test.SharedSQLContext
-import org.apache.spark.sql.types.{NullType, StructType}
+import org.apache.spark.sql.types._
 import org.apache.spark.storage.StorageLevel
 
 
@@ -547,9 +547,28 @@ class CatalogSuite
     assert(spark.table("my_temp_table").storageLevel == StorageLevel.DISK_ONLY)
   }
 
-  test("SPARK-28443: Spark sql add exception when create field type NullType") {
+  test("SPARK-28443: fail the DDL command if it creates a table with null-type columns") {
+    val schema1 = new StructType().add("c", NullType)
+    checkTableNullType(schema1)
+
+    val schema2 = new StructType()
+      .add("c", StructType(Seq(StructField.apply("c1", NullType))))
+    checkTableNullType(schema2)
+
+    val schema3 = new StructType().add("c", ArrayType(NullType))
+    checkTableNullType(schema3)
+
+    val schema4 = new StructType()
+      .add("c", MapType(StringType, NullType))
+    checkTableNullType(schema4)
+
+    val schema5 = new StructType()
+      .add("c", MapType(NullType, StringType))
+    checkTableNullType(schema5)
+  }
+
+  private def checkTableNullType(schema: StructType): Unit = {
     val expectedMsg = "DataType NullType is not supported for create table"
-    val schema = new StructType().add("c", NullType)
     withTable("t") {
       val e = intercept[AnalysisException] {
         spark.catalog.createTable(