[SPARK-32356][SQL] Forbid create view with null type

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala

@@ -356,7 +356,7 @@ private[sql] object CatalogV2Util {
     }
     if (containsNullType(dt)) {
       throw new AnalysisException(
-        s"Cannot create tables with ${NullType.simpleString} type.")
+        s"Cannot create tables/views with ${NullType.simpleString} type.")
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -529,6 +529,9 @@ class ResolveSessionCatalog(
         partitionSpec)
 
     case AlterViewAsStatement(name, originalText, query) =>
+      if (query.resolved) {
+        assertNoNullTypeInSchema(query.schema)
+      }
       val viewName = parseTempViewOrV1Table(name, "ALTER VIEW QUERY")
       AlterViewAsCommand(
         viewName.asTableIdentifier,
@@ -538,7 +541,9 @@ class ResolveSessionCatalog(
     case CreateViewStatement(
       tbl, userSpecifiedColumns, comment, properties,
       originalText, child, allowExisting, replace, viewType) =>
-
+      if (child.resolved) {
+        assertNoNullTypeInSchema(child.schema)
+      }
       val v1TableName = if (viewType != PersistedView) {
         // temp view doesn't belong to any catalog and we shouldn't resolve catalog in the name.
         tbl

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

@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeRef
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project, View}
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.NamespaceHelper
+import org.apache.spark.sql.connector.catalog.CatalogV2Util.assertNoNullTypeInSchema
 import org.apache.spark.sql.internal.StaticSQLConf
 import org.apache.spark.sql.types.{BooleanType, MetadataBuilder, StringType}
 import org.apache.spark.sql.util.SchemaUtils
@@ -96,6 +97,7 @@ case class CreateViewCommand(
     qe.assertAnalyzed()
     val analyzedPlan = qe.analyzed
 
+    assertNoNullTypeInSchema(analyzedPlan.schema)
     if (userSpecifiedColumns.nonEmpty &&
         userSpecifiedColumns.length != analyzedPlan.output.length) {
       throw new AnalysisException(s"The number of columns produced by the SELECT clause " +

sql/core/src/test/scala/org/apache/spark/sql/execution/SQLViewSuite.scala

@@ -22,6 +22,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
 import org.apache.spark.sql.internal.SQLConf.MAX_NESTED_VIEW_DEPTH
 import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils}
+import org.apache.spark.sql.types.NullType
 
 class SimpleSQLViewSuite extends SQLViewSuite with SharedSparkSession
 
@@ -735,4 +736,30 @@ abstract class SQLViewSuite extends QueryTest with SQLTestUtils {
       }
     }
   }
+
+  test("SPARK-32356: Forbid create view with null type in sql side") {
+    // test sql
+    val sql1 = "create view v as select null as c"
+    val sql2 = "alter view v as select null as c"
+    val sql3 = "create temporary view v as select null as c"
+    val sql4 = "create global temporary view v as select null as c"
+    Seq(sql1, sql2, sql3, sql4).foreach { input =>
+      val msg = intercept[AnalysisException] {
+        sql(input)
+      }.getMessage
+      assert(msg.contains(s"Cannot create tables/views with ${NullType.simpleString} type."))
+    }
+  }
+
+  test("SPARK-32356: Forbid create view with null type in dataset side") {
+    // test df.createTempView
+    val msg = intercept[AnalysisException] {
+      sql("select null as c").createTempView("null_type_view")
+    }.getMessage
+    assert(msg.contains(s"Cannot create tables/views with ${NullType.simpleString} type."))
+    val msg2 = intercept[AnalysisException] {
+      sql("select null as c").createGlobalTempView("null_type_view")
+    }.getMessage
+    assert(msg2.contains(s"Cannot create tables/views with ${NullType.simpleString} type."))
+  }
 }

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -2314,6 +2314,7 @@ class HiveDDLSuite
   }
 
   test("SPARK-20680: do not support for null column datatype") {
+    val errMsg = s"Cannot create tables/views with ${NullType.simpleString} type."
     withTable("t") {
       withView("tabNullType") {
         hiveClient.runSqlHive("CREATE TABLE t (t1 int)")
@@ -2331,17 +2332,17 @@ class HiveDDLSuite
       val e1 = intercept[AnalysisException] {
         spark.sql("CREATE TABLE t1 USING PARQUET AS SELECT null as null_col")
       }.getMessage
-      assert(e1.contains("Cannot create tables with null type"))
+      assert(e1.contains(errMsg))
 
       val e2 = intercept[AnalysisException] {
         spark.sql("CREATE TABLE t2 AS SELECT null as null_col")
       }.getMessage
-      assert(e2.contains("Cannot create tables with null type"))
+      assert(e2.contains(errMsg))
 
       val e3 = intercept[AnalysisException] {
         spark.sql("CREATE TABLE t3 STORED AS PARQUET AS SELECT null as null_col")
       }.getMessage
-      assert(e3.contains("Cannot create tables with null type"))
+      assert(e3.contains(errMsg))
     }
 
     // Forbid Replace table AS SELECT with null type
@@ -2350,27 +2351,27 @@ class HiveDDLSuite
       val e = intercept[AnalysisException] {
         spark.sql(s"CREATE OR REPLACE TABLE t USING $v2Source AS SELECT null as null_col")
       }.getMessage
-      assert(e.contains("Cannot create tables with null type"))
+      assert(e.contains(errMsg))
     }
 
     // Forbid creating table with VOID type in Spark
     withTable("t1", "t2", "t3", "t4") {
       val e1 = intercept[AnalysisException] {
         spark.sql(s"CREATE TABLE t1 (v VOID) USING PARQUET")
       }.getMessage
-      assert(e1.contains("Cannot create tables with null type"))
+      assert(e1.contains(errMsg))
       val e2 = intercept[AnalysisException] {
         spark.sql(s"CREATE TABLE t2 (v VOID) USING hive")
       }.getMessage
-      assert(e2.contains("Cannot create tables with null type"))
+      assert(e2.contains(errMsg))
       val e3 = intercept[AnalysisException] {
         spark.sql(s"CREATE TABLE t3 (v VOID)")
       }.getMessage
-      assert(e3.contains("Cannot create tables with null type"))
+      assert(e3.contains(errMsg))
       val e4 = intercept[AnalysisException] {
         spark.sql(s"CREATE TABLE t4 (v VOID) STORED AS PARQUET")
       }.getMessage
-      assert(e4.contains("Cannot create tables with null type"))
+      assert(e4.contains(errMsg))
     }
 
     // Forbid Replace table with VOID type
@@ -2379,7 +2380,7 @@ class HiveDDLSuite
       val e = intercept[AnalysisException] {
         spark.sql(s"CREATE OR REPLACE TABLE t (v VOID) USING $v2Source")
       }.getMessage
-      assert(e.contains("Cannot create tables with null type"))
+      assert(e.contains(errMsg))
     }
 
     // Make sure spark.catalog.createTable with null type will fail
@@ -2416,7 +2417,7 @@ class HiveDDLSuite
           schema = schema,
           options = Map("fileFormat" -> "parquet"))
       }.getMessage
-      assert(e.contains("Cannot create tables with null type"))
+      assert(e.contains(s"Cannot create tables/views with ${NullType.simpleString} type."))
     }
   }
 
@@ -2429,7 +2430,7 @@ class HiveDDLSuite
           schema = schema,
           options = Map.empty[String, String])
       }.getMessage
-      assert(e.contains("Cannot create tables with null type"))
+      assert(e.contains(s"Cannot create tables/views with ${NullType.simpleString} type."))
     }
   }