[SPARK-21529][SQL][4.X] Improve the error message for unsupported Hive union type

common/utils/src/main/resources/error/error-conditions.json

@@ -8641,6 +8641,12 @@
     ],
     "sqlState" : "42K0E"
   },
+  "UNSUPPORTED_HIVE_TYPE" : {
+    "message" : [
+      "Cannot read the Hive type <fieldType> of the column <fieldName> because Spark SQL does not support this data type."
+    ],
+    "sqlState" : "0A000"
+  },
   "UNSUPPORTED_INSERT" : {
     "message" : [
       "Can't insert into the target."

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -1696,6 +1696,13 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       cause = e)
   }
 
+  def unsupportedHiveTypeError(fieldType: String, fieldName: String): Throwable = {
+    new SparkUnsupportedOperationException(
+      errorClass = "UNSUPPORTED_HIVE_TYPE",
+      messageParameters = Map(
+        "fieldType" -> toSQLType(fieldType),
+        "fieldName" -> toSQLId(fieldName)))
+  }
   def getTablesByTypeUnsupportedByHiveVersionError(): SparkUnsupportedOperationException = {
     new SparkUnsupportedOperationException(
       errorClass = "GET_TABLES_BY_TYPE_UNSUPPORTED_BY_HIVE_VERSION")

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -1139,6 +1139,11 @@ private[hive] object HiveClientImpl extends Logging {
       CatalystSqlParser.parseDataType(typeStr)
     } catch {
       case e: ParseException =>
+        // Hive's union type (uniontype<...>) is not supported by Spark SQL and makes the parser
+        // fail with a generic message. Detect it and report a clearer error (SPARK-21529).
+        if (hc.getType.toLowerCase(Locale.ROOT).contains("uniontype<")) {
+          throw QueryExecutionErrors.unsupportedHiveTypeError(hc.getType, hc.getName)
+        }
         throw QueryExecutionErrors.cannotRecognizeHiveTypeError(e, typeStr, hc.getName)
     }
   }

sql/hive/src/test/scala/org/apache/spark/sql/hive/client/HiveClientImplSuite.scala

@@ -0,0 +1,49 @@
+/*
+ * 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.hive.client
+
+import org.apache.hadoop.hive.metastore.api.FieldSchema
+
+import org.apache.spark.{SparkFunSuite, SparkUnsupportedOperationException}
+
+class HiveClientImplSuite extends SparkFunSuite {
+
+  test("SPARK-21529: a clear error is raised for an unsupported Hive union type") {
+    val column = new FieldSchema("c", "uniontype<int,string>", null)
+    checkError(
+      exception = intercept[SparkUnsupportedOperationException] {
+        HiveClientImpl.fromHiveColumn(column)
+      },
+      condition = "UNSUPPORTED_HIVE_TYPE",
+      parameters = Map(
+        "fieldType" -> "\"UNIONTYPE<INT,STRING>\"",
+        "fieldName" -> "`c`"))
+  }
+
+  test("SPARK-21529: a Hive union type nested in a struct is detected") {
+    val column = new FieldSchema("c", "struct<a:uniontype<int,string>>", null)
+    checkError(
+      exception = intercept[SparkUnsupportedOperationException] {
+        HiveClientImpl.fromHiveColumn(column)
+      },
+      condition = "UNSUPPORTED_HIVE_TYPE",
+      parameters = Map(
+        "fieldType" -> "\"STRUCT<A:UNIONTYPE<INT,STRING>>\"",
+        "fieldName" -> "`c`"))
+  }
+}