[SPARK-20680][SQL] Adding HiveVoidType in Spark to be compatible with Hive

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -2184,7 +2184,9 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
    * Create a Spark DataType.
    */
   private def visitSparkDataType(ctx: DataTypeContext): DataType = {
-    HiveStringType.replaceCharType(typedVisit(ctx))
+    HiveNullType.replaceNullType(
+      HiveStringType.replaceCharType(typedVisit(ctx))
+    )
   }
 
   /**
@@ -2212,6 +2214,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
       case ("decimal" | "dec" | "numeric", precision :: scale :: Nil) =>
         DecimalType(precision.getText.toInt, scale.getText.toInt)
       case ("interval", Nil) => CalendarIntervalType
+      case ("void", Nil) => HiveNullType
       case (dt, params) =>
         val dtStr = if (params.nonEmpty) s"$dt(${params.mkString(",")})" else dt
         throw new ParseException(s"DataType $dtStr is not supported.", ctx)
@@ -2260,7 +2263,9 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
 
     // Add Hive type string to metadata.
     val rawDataType = typedVisit[DataType](ctx.dataType)
-    val cleanedDataType = HiveStringType.replaceCharType(rawDataType)
+    val cleanedDataType = HiveNullType.replaceNullType(
+      HiveStringType.replaceCharType(rawDataType)
+    )
     if (rawDataType != cleanedDataType) {
       builder.putString(HIVE_TYPE_STRING, rawDataType.catalogString)
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/types/HiveNullType.scala

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.types
+
+/**
+ * A hive null type for compatibility. These datatypes should only used for parsing,
+ * and should NOT be used anywhere else. Any instance of these data types should be
+ * replaced by a [[NullType]] before analysis.
+ */
+class HiveNullType private() extends DataType {
+
+  override def defaultSize: Int = 1
+
+  override private[spark] def asNullable: HiveNullType = this
+
+  override def simpleString: String = "void"
+}
+
+case object HiveNullType extends HiveNullType {
+  def replaceNullType(dt: DataType): DataType = dt match {
+    case ArrayType(et, nullable) =>
+      ArrayType(replaceNullType(et), nullable)
+    case MapType(kt, vt, nullable) =>
+      MapType(replaceNullType(kt), replaceNullType(vt), nullable)
+    case StructType(fields) =>
+      StructType(fields.map { field =>
+        field.copy(dataType = replaceNullType(field.dataType))
+      })
+    case _: HiveNullType => NullType
+    case _ => dt
+  }
+
+
+  def containsNullType(dt: DataType): Boolean = dt match {
+    case ArrayType(et, _) => containsNullType(et)
+    case MapType(kt, vt, _) => containsNullType(kt) || containsNullType(vt)
+    case StructType(fields) => fields.exists(f => containsNullType(f.dataType))
+    case _ => dt.isInstanceOf[HiveNullType]
+  }
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DataTypeParserSuite.scala

@@ -62,6 +62,7 @@ class DataTypeParserSuite extends SparkFunSuite {
   checkDataType("cHaR(27)", StringType)
   checkDataType("BINARY", BinaryType)
   checkDataType("interval", CalendarIntervalType)
+  checkDataType("void", NullType)
 
   checkDataType("array<doublE>", ArrayType(DoubleType, true))
   checkDataType("Array<map<int, tinYint>>", ArrayType(MapType(IntegerType, ByteType, true), true))

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

@@ -28,7 +28,7 @@ import org.apache.spark.sql.execution.command._
 import org.apache.spark.sql.execution.datasources.{CreateTable, DataSource, RefreshTable}
 import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{HIVE_TYPE_STRING, HiveStringType, MetadataBuilder, StructField, StructType}
+import org.apache.spark.sql.types.{HIVE_TYPE_STRING, HiveNullType, HiveStringType, MetadataBuilder, StructField, StructType}
 
 /**
  * Resolves catalogs from the multi-part identifiers in SQL statements, and convert the statements
@@ -132,7 +132,9 @@ class ResolveSessionCatalog(
               }
           }
           // Add Hive type string to metadata.
-          val cleanedDataType = HiveStringType.replaceCharType(dataType)
+          val cleanedDataType = HiveNullType.replaceNullType(
+            HiveStringType.replaceCharType(dataType)
+          )
           if (dataType != cleanedDataType) {
             builder.putString(HIVE_TYPE_STRING, dataType.catalogString)
           }
@@ -719,7 +721,9 @@ class ResolveSessionCatalog(
     val builder = new MetadataBuilder
     col.comment.foreach(builder.putString("comment", _))
 
-    val cleanedDataType = HiveStringType.replaceCharType(col.dataType)
+    val cleanedDataType = HiveNullType.replaceNullType(
+      HiveStringType.replaceCharType(col.dataType)
+    )
     if (col.dataType != cleanedDataType) {
       builder.putString(HIVE_TYPE_STRING, col.dataType.catalogString)
     }

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

@@ -2309,6 +2309,22 @@ class HiveDDLSuite
     }
   }
 
+  test("SPARK-20680: Spark-sql do not support for void column datatype of view") {
+    withTable("t") {
+      withView("tabNullType") {
+        val client =
+          spark.sharedState.externalCatalog.unwrapped.asInstanceOf[HiveExternalCatalog].client
+        client.runSqlHive("CREATE TABLE t (t1 int)")
+        client.runSqlHive("INSERT INTO t VALUES (3)")
+        client.runSqlHive("CREATE VIEW tabNullType AS SELECT NULL AS col FROM t")
+        checkAnswer(spark.table("tabNullType"), Row(null))
+        // No exception shows
+        val desc = spark.sql("DESC tabNullType").collect().toSeq
+        assert(desc.contains(Row("col", "null", null)))
+      }
+    }
+  }
+
   test("SPARK-21216: join with a streaming DataFrame") {
     import org.apache.spark.sql.execution.streaming.MemoryStream
     import testImplicits._