Simplify the logic and throw meaningful exception message.

Author: dongjoon-hyun

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -283,12 +283,10 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
     } else {
       val partitionSpec = if (ctx.partitionSpec != null) {
         // According to the syntax, visitPartitionSpec returns `Map[String, Option[String]]`.
-        val (valid, nonValid) = visitPartitionSpec(ctx.partitionSpec).partition(_._2.isDefined)
-        if (nonValid.nonEmpty) {
-          // For non-valid specification for `DESC` command, this raises a ParseException.
-          return null
-        } else {
-          valid.map(x => (x._1, x._2.get))
+        visitPartitionSpec(ctx.partitionSpec).map {
+          case (key, Some(value)) => key -> value
+          case (key, _) =>
+            throw new ParseException(s"PARTITION specification is incomplete: `$key`", ctx)
         }
       } else {
         Map.empty[String, String]

sql/core/src/test/resources/sql-tests/inputs/describe.sql

@@ -20,5 +20,5 @@ DESC t PARTITION (c='Us', d=2);
 -- AnalysisException: Partition spec is invalid
 DESC t PARTITION (c='Us');
 
--- ParseException: Unsupported SQL statement
+-- ParseException: PARTITION specification is incomplete
 DESC t PARTITION (c='Us', d);

sql/core/src/test/resources/sql-tests/results/describe.sql.out

@@ -75,6 +75,8 @@ struct<>
 -- !query 6 output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-Unsupported SQL statement
+PARTITION specification is incomplete: `d`(line 1, pos 0)
+
 == SQL ==
 DESC t PARTITION (c='Us', d)
+^^^

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

@@ -389,7 +389,7 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       val m3 = intercept[ParseException] {
         sql("DESC partitioned_table PARTITION (c='Us', d)")
       }.getMessage()
-      assert(m3.contains("Unsupported SQL statement"))
+      assert(m3.contains("PARTITION specification is incomplete: `d`"))
 
       spark
         .range(1).select('id as 'a, 'id as 'b, 'id as 'c, 'id as 'd).write