From 23dcceac46947a1bb805e852cc62eb9538cadaee Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Sat, 10 Jun 2017 12:16:45 -0700
Subject: [PATCH 01/12] support desc table column

---
 .../spark/sql/execution/SparkSqlParser.scala  |  11 +-
 .../spark/sql/execution/command/tables.scala  | 103 ++++++++++++++++++
 .../org/apache/spark/sql/SQLQuerySuite.scala  |  38 +++++++
 .../spark/sql/StatisticsCollectionSuite.scala |  23 +++-
 .../sql/execution/SparkSqlParserSuite.scala   |  15 +++
 5 files changed, 186 insertions(+), 4 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 3c58c6e1b6780..3aab8496f4eb8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -306,10 +306,15 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
    * Create a [[DescribeTableCommand]] logical plan.
    */
   override def visitDescribeTable(ctx: DescribeTableContext): LogicalPlan = withOrigin(ctx) {
-    // Describe column are not supported yet. Return null and let the parser decide
-    // what to do with this (create an exception or pass it on to a different system).
     if (ctx.describeColName != null) {
-      null
+      if (ctx.partitionSpec != null) {
+        throw new ParseException("DESC TABLE COLUMN for a specific partition is not supported", ctx)
+      } else {
+        DescribeColumnCommand(
+          visitTableIdentifier(ctx.tableIdentifier),
+          ctx.describeColName.getText,
+          ctx.FORMATTED != null)
+      }
     } else {
       val partitionSpec = if (ctx.partitionSpec != null) {
         // According to the syntax, visitPartitionSpec returns `Map[String, Option[String]]`.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index b937a8a9f375b..872ce2cfa0f5c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -36,6 +36,7 @@ import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.CatalogTableType._
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
+import org.apache.spark.sql.catalyst.plans.logical.ColumnStat
 import org.apache.spark.sql.catalyst.util.quoteIdentifier
 import org.apache.spark.sql.execution.datasources.{DataSource, FileFormat, PartitioningUtils}
 import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat
@@ -619,6 +620,108 @@ case class DescribeTableCommand(
   }
 }
 
+/**
+ * A command to list the info for a column, including name, data type, column stats and comment.
+ * This function creates a [[DescribeColumnCommand]] logical plan.
+ *
+ * The syntax of using this command in SQL is:
+ * {{{
+ *   DESCRIBE [EXTENDED|FORMATTED] table_name column_name;
+ * }}}
+ */
+case class DescribeColumnCommand(
+    table: TableIdentifier,
+    column: String,
+    isFormatted: Boolean)
+  extends RunnableCommand {
+
+  override val output: Seq[Attribute] = {
+    // The displayed names are based on Hive.
+    // (Link for the corresponding Hive Jira: https://issues.apache.org/jira/browse/HIVE-7050)
+    if (isFormatted) {
+      Seq(
+        AttributeReference("col_name", StringType, nullable = false,
+          new MetadataBuilder().putString("comment", "name of the column").build())(),
+        AttributeReference("data_type", StringType, nullable = false,
+          new MetadataBuilder().putString("comment", "data type of the column").build())(),
+        AttributeReference("min", StringType, nullable = true,
+          new MetadataBuilder().putString("comment", "min value of the column").build())(),
+        AttributeReference("max", StringType, nullable = true,
+          new MetadataBuilder().putString("comment", "max value of the column").build())(),
+        AttributeReference("num_nulls", StringType, nullable = true,
+          new MetadataBuilder().putString("comment", "number of nulls of the column").build())(),
+        AttributeReference("distinct_count", StringType, nullable = true,
+          new MetadataBuilder().putString("comment", "distinct count of the column").build())(),
+        AttributeReference("avg_col_len", StringType, nullable = true,
+          new MetadataBuilder().putString("comment",
+            "average length of the values of the column").build())(),
+        AttributeReference("max_col_len", StringType, nullable = true,
+          new MetadataBuilder().putString("comment",
+            "max length of the values of the column").build())(),
+        AttributeReference("comment", StringType, nullable = true,
+          new MetadataBuilder().putString("comment", "comment of the column").build())())
+    } else {
+      Seq(
+        AttributeReference("col_name", StringType, nullable = false,
+          new MetadataBuilder().putString("comment", "name of the column").build())(),
+        AttributeReference("data_type", StringType, nullable = false,
+          new MetadataBuilder().putString("comment", "data type of the column").build())(),
+        AttributeReference("comment", StringType, nullable = true,
+          new MetadataBuilder().putString("comment", "comment of the column").build())())
+    }
+  }
+
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    val catalog = sparkSession.sessionState.catalog
+    val resolver = sparkSession.sessionState.conf.resolver
+    val catalogTable = catalog.getTempViewOrPermanentTableMetadata(table)
+    val attribute = {
+      val field = catalogTable.schema.find(f => resolver(f.name, column))
+      field.getOrElse {
+        if (column.contains(".")) {
+          throw new AnalysisException(
+            s"DESC TABLE COLUMN is not supported for nested column: $column")
+        } else {
+          throw new AnalysisException(s"Column $column does not exist.")
+        }
+      }
+    }
+
+    val colStats: Map[String, ColumnStat] = if (catalog.isTemporaryTable(table)) {
+      Map.empty
+    } else {
+      catalogTable.stats.map(_.colStats).getOrElse(Map.empty)
+    }
+    val cs = colStats.get(attribute.name)
+
+    val comment = if (attribute.metadata.contains("comment")) {
+      Option(attribute.metadata.getString("comment"))
+    } else {
+      None
+    }
+
+    val result = if (isFormatted) {
+      // Show column stats only when formatted is specified
+      Row(
+        attribute.name,
+        attribute.dataType.simpleString,
+        cs.flatMap(_.min.map(_.toString)).orNull,
+        cs.flatMap(_.max.map(_.toString)).orNull,
+        cs.map(_.nullCount.toString).orNull,
+        cs.map(_.distinctCount.toString).orNull,
+        cs.map(_.avgLen.toString).orNull,
+        cs.map(_.maxLen.toString).orNull,
+        comment.orNull)
+    } else {
+      Row(
+        attribute.name,
+        attribute.dataType.simpleString,
+        comment.orNull)
+    }
+
+    Seq(result)
+  }
+}
 
 /**
  * A command for users to get tables in the given database.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index 68f61cfab6d2f..d2a4320324bee 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -39,6 +39,44 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
 
   setupTestData()
 
+  test("describe table column") {
+    def checkDescColumn(
+        table: String,
+        column: String,
+        dataType: String,
+        comment: Option[String]): Unit = {
+      checkAnswer(sql(s"desc $table $column"), Row(column, dataType, comment.orNull))
+      checkAnswer(sql(s"desc extended $table $column"), Row(column, dataType, comment.orNull))
+      checkAnswer(sql(s"desc formatted $table $column"),
+        Row(column, dataType, null, null, null, null, null, null, comment.orNull))
+    }
+
+    val comment = "foo bar"
+    // Test temp table
+    withTempView("desc_col_tempTable") {
+      sql(s"create temporary view desc_col_tempTable (key int comment '$comment') using parquet")
+      checkDescColumn("desc_col_tempTable", "key", "int", Some(comment))
+
+      // Describe a non-existent column
+      val msg = intercept[AnalysisException](sql("desc desc_col_tempTable key1")).getMessage
+      assert(msg.contains("Column key1 does not exist."))
+    }
+
+    withTable("desc_col_persistentTable", "desc_col_complexTable") {
+      // Test persistent table
+      sql(s"create table desc_col_persistentTable (key int comment '$comment') using parquet")
+      checkDescColumn("desc_col_persistentTable", "key", "int", Some(comment))
+
+      // Test complex column
+      complexData.write.saveAsTable("desc_col_complexTable")
+      checkDescColumn("desc_col_complexTable", "s", "struct<key:int,value:string>", None)
+
+      // Describe a nested column
+      val msg = intercept[AnalysisException](sql("desc desc_col_complexTable s.key")).getMessage
+      assert(msg.contains("DESC TABLE COLUMN is not supported for nested column: s.key"))
+    }
+  }
+
   test("SPARK-8010: promote numeric to string") {
     val df = Seq((1, 1)).toDF("key", "value")
     df.createOrReplaceTempView("src")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
index 9824062f969b3..d37eb52132993 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
@@ -158,7 +158,7 @@ class StatisticsCollectionSuite extends StatisticsCollectionTestBase with Shared
       BigInt(4444444444L) -> ("4.1 GB", "4.44E+9"),
       BigInt(5555555555555L) -> ("5.1 TB", "5.56E+12"),
       BigInt(6666666666666666L) -> ("5.9 PB", "6.67E+15"),
-      BigInt(1L << 10 ) * (1L << 60) -> ("1024.0 EB", "1.18E+21"),
+      BigInt(1L << 10) * (1L << 60) -> ("1024.0 EB", "1.18E+21"),
       BigInt(1L << 11) * (1L << 60) -> ("2.36E+21 B", "2.36E+21")
     )
     numbers.foreach { case (input, (expectedSize, expectedRows)) =>
@@ -168,6 +168,27 @@ class StatisticsCollectionSuite extends StatisticsCollectionTestBase with Shared
       assert(stats.simpleString == expectedString)
     }
   }
+
+  test("desc column with stats") {
+    withTable("desc_col_statsTable") {
+      val columns = stats.keys.toSeq
+      val df = data.toDF(columns :+ "carray" : _*)
+      df.write.saveAsTable("desc_col_statsTable")
+      sql(s"analyze table desc_col_statsTable compute statistics for columns " +
+        s"${columns.mkString(", ")}")
+      stats.zip(df.schema).foreach { case ((colName, columnStat), structField) =>
+        checkAnswer(sql(s"desc formatted desc_col_statsTable ${structField.name}"),
+          Row(colName, structField.dataType.simpleString,
+            columnStat.min.map(_.toString).orNull,
+            columnStat.max.map(_.toString).orNull,
+            columnStat.nullCount.toString,
+            columnStat.distinctCount.toString,
+            columnStat.avgLen.toString,
+            columnStat.maxLen.toString,
+            structField.getComment().orNull))
+      }
+    }
+  }
 }
 
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
index bd9c2ebd6fab9..6636e279dcfec 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
@@ -234,6 +234,21 @@ class SparkSqlParserSuite extends AnalysisTest {
     intercept("explain describe tables x", "Unsupported SQL statement")
   }
 
+  test("describe table column") {
+    assertEqual("describe table t col",
+      DescribeColumnCommand(
+        TableIdentifier("t"), "col", isFormatted = false))
+    assertEqual("describe table extended t col",
+      DescribeColumnCommand(
+        TableIdentifier("t"), "col", isFormatted = false))
+    assertEqual("describe table formatted t col",
+      DescribeColumnCommand(
+        TableIdentifier("t"), "col", isFormatted = true))
+
+    intercept("describe table t partition (ds='1970-01-01') col",
+      "DESC TABLE COLUMN for a specific partition is not supported")
+  }
+
   test("analyze table statistics") {
     assertEqual("analyze table t compute statistics",
       AnalyzeTableCommand(TableIdentifier("t"), noscan = false))

From 9ac8c11a667657369d4b21e42e5bc9996042c270 Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Sun, 11 Jun 2017 12:36:45 -0700
Subject: [PATCH 02/12] upper case for command keywords

---
 .../scala/org/apache/spark/sql/SQLQuerySuite.scala | 14 +++++++-------
 .../spark/sql/StatisticsCollectionSuite.scala      |  4 ++--
 .../spark/sql/execution/SparkSqlParserSuite.scala  | 11 +++++++----
 3 files changed, 16 insertions(+), 13 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index d2a4320324bee..3c5e16b935971 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -45,26 +45,26 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
         column: String,
         dataType: String,
         comment: Option[String]): Unit = {
-      checkAnswer(sql(s"desc $table $column"), Row(column, dataType, comment.orNull))
-      checkAnswer(sql(s"desc extended $table $column"), Row(column, dataType, comment.orNull))
-      checkAnswer(sql(s"desc formatted $table $column"),
+      checkAnswer(sql(s"DESC $table $column"), Row(column, dataType, comment.orNull))
+      checkAnswer(sql(s"DESC EXTENDED $table $column"), Row(column, dataType, comment.orNull))
+      checkAnswer(sql(s"DESC FORMATTED $table $column"),
         Row(column, dataType, null, null, null, null, null, null, comment.orNull))
     }
 
     val comment = "foo bar"
     // Test temp table
     withTempView("desc_col_tempTable") {
-      sql(s"create temporary view desc_col_tempTable (key int comment '$comment') using parquet")
+      sql(s"CREATE TEMPORARY VIEW desc_col_tempTable (key int COMMENT '$comment') USING PARQUET")
       checkDescColumn("desc_col_tempTable", "key", "int", Some(comment))
 
       // Describe a non-existent column
-      val msg = intercept[AnalysisException](sql("desc desc_col_tempTable key1")).getMessage
+      val msg = intercept[AnalysisException](sql("DESC desc_col_tempTable key1")).getMessage
       assert(msg.contains("Column key1 does not exist."))
     }
 
     withTable("desc_col_persistentTable", "desc_col_complexTable") {
       // Test persistent table
-      sql(s"create table desc_col_persistentTable (key int comment '$comment') using parquet")
+      sql(s"CREATE TABLE desc_col_persistentTable (key int COMMENT '$comment') USING PARQUET")
       checkDescColumn("desc_col_persistentTable", "key", "int", Some(comment))
 
       // Test complex column
@@ -72,7 +72,7 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
       checkDescColumn("desc_col_complexTable", "s", "struct<key:int,value:string>", None)
 
       // Describe a nested column
-      val msg = intercept[AnalysisException](sql("desc desc_col_complexTable s.key")).getMessage
+      val msg = intercept[AnalysisException](sql("DESC desc_col_complexTable s.key")).getMessage
       assert(msg.contains("DESC TABLE COLUMN is not supported for nested column: s.key"))
     }
   }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
index d37eb52132993..ad2866b5e0637 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
@@ -174,10 +174,10 @@ class StatisticsCollectionSuite extends StatisticsCollectionTestBase with Shared
       val columns = stats.keys.toSeq
       val df = data.toDF(columns :+ "carray" : _*)
       df.write.saveAsTable("desc_col_statsTable")
-      sql(s"analyze table desc_col_statsTable compute statistics for columns " +
+      sql(s"ANALYZE TABLE desc_col_statsTable COMPUTE STATISTICS FOR COLUMNS " +
         s"${columns.mkString(", ")}")
       stats.zip(df.schema).foreach { case ((colName, columnStat), structField) =>
-        checkAnswer(sql(s"desc formatted desc_col_statsTable ${structField.name}"),
+        checkAnswer(sql(s"DESC FORMATTED desc_col_statsTable ${structField.name}"),
           Row(colName, structField.dataType.simpleString,
             columnStat.min.map(_.toString).orNull,
             columnStat.max.map(_.toString).orNull,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
index 6636e279dcfec..3306ad1cfceb1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
@@ -235,17 +235,20 @@ class SparkSqlParserSuite extends AnalysisTest {
   }
 
   test("describe table column") {
-    assertEqual("describe table t col",
+    assertEqual("DESCRIBE t col",
       DescribeColumnCommand(
         TableIdentifier("t"), "col", isFormatted = false))
-    assertEqual("describe table extended t col",
+    assertEqual("DESCRIBE TABLE t col",
       DescribeColumnCommand(
         TableIdentifier("t"), "col", isFormatted = false))
-    assertEqual("describe table formatted t col",
+    assertEqual("DESCRIBE TABLE EXTENDED t col",
+      DescribeColumnCommand(
+        TableIdentifier("t"), "col", isFormatted = false))
+    assertEqual("DESCRIBE TABLE FORMATTED t col",
       DescribeColumnCommand(
         TableIdentifier("t"), "col", isFormatted = true))
 
-    intercept("describe table t partition (ds='1970-01-01') col",
+    intercept("DESCRIBE TABLE t PARTITION (ds='1970-01-01') col",
       "DESC TABLE COLUMN for a specific partition is not supported")
   }
 

From 16c5219ba3e63ab2174e66315539b12d2d735aaf Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Sun, 11 Jun 2017 23:51:35 -0700
Subject: [PATCH 03/12] fix test

---
 .../org/apache/spark/sql/execution/SparkSqlParserSuite.scala    | 2 --
 1 file changed, 2 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
index 3306ad1cfceb1..bc2674c8d6597 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
@@ -230,8 +230,6 @@ class SparkSqlParserSuite extends AnalysisTest {
     assertEqual("describe table formatted t",
       DescribeTableCommand(
         TableIdentifier("t"), Map.empty, isExtended = true))
-
-    intercept("explain describe tables x", "Unsupported SQL statement")
   }
 
   test("describe table column") {

From 21e7b37bf2d9dd07b19925c17f4d8c77282e81bc Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Mon, 12 Jun 2017 10:43:29 -0700
Subject: [PATCH 04/12] minor improve

---
 .../apache/spark/sql/execution/command/tables.scala    | 10 +++-------
 1 file changed, 3 insertions(+), 7 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 872ce2cfa0f5c..60a8ca9c77f0a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -657,7 +657,7 @@ case class DescribeColumnCommand(
             "average length of the values of the column").build())(),
         AttributeReference("max_col_len", StringType, nullable = true,
           new MetadataBuilder().putString("comment",
-            "max length of the values of the column").build())(),
+            "maximum length of the values of the column").build())(),
         AttributeReference("comment", StringType, nullable = true,
           new MetadataBuilder().putString("comment", "comment of the column").build())())
     } else {
@@ -687,11 +687,7 @@ case class DescribeColumnCommand(
       }
     }
 
-    val colStats: Map[String, ColumnStat] = if (catalog.isTemporaryTable(table)) {
-      Map.empty
-    } else {
-      catalogTable.stats.map(_.colStats).getOrElse(Map.empty)
-    }
+    val colStats = catalogTable.stats.map(_.colStats).getOrElse(Map.empty)
     val cs = colStats.get(attribute.name)
 
     val comment = if (attribute.metadata.contains("comment")) {
@@ -701,7 +697,7 @@ case class DescribeColumnCommand(
     }
 
     val result = if (isFormatted) {
-      // Show column stats only when formatted is specified
+      // Show column stats only when formatted is specified.
       Row(
         attribute.name,
         attribute.dataType.simpleString,

From 7c901ceecc74eb8b3bf1fc10e60a79401b30367b Mon Sep 17 00:00:00 2001
From: wangzhenhua <wangzhenhua@huawei.com>
Date: Sat, 24 Jun 2017 15:07:37 +0800
Subject: [PATCH 05/12] 1. change sql syntax to support complex name but not
 nested column; 2. use catalogString instead of simpleString

---
 .../apache/spark/sql/catalyst/parser/SqlBase.g4  |  2 +-
 .../spark/sql/execution/command/tables.scala     | 16 ++++------------
 .../org/apache/spark/sql/SQLQuerySuite.scala     |  4 ----
 .../sql/execution/SparkSqlParserSuite.scala      |  6 ++++++
 4 files changed, 11 insertions(+), 17 deletions(-)

diff --git a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
index 9456031736528..89186ccaa00d6 100644
--- a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
+++ b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
@@ -262,7 +262,7 @@ describeFuncName
     ;
 
 describeColName
-    : identifier ('.' (identifier | STRING))*
+    : identifier
     ;
 
 ctes
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 60a8ca9c77f0a..c0b1439015240 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -20,13 +20,11 @@ package org.apache.spark.sql.execution.command
 import java.io.File
 import java.net.URI
 import java.nio.file.FileSystems
-import java.util.Date
 
 import scala.collection.mutable.ArrayBuffer
 import scala.util.control.NonFatal
 import scala.util.Try
 
-import org.apache.commons.lang3.StringEscapeUtils
 import org.apache.hadoop.fs.Path
 
 import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
@@ -36,9 +34,8 @@ import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.CatalogTableType._
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
-import org.apache.spark.sql.catalyst.plans.logical.ColumnStat
 import org.apache.spark.sql.catalyst.util.quoteIdentifier
-import org.apache.spark.sql.execution.datasources.{DataSource, FileFormat, PartitioningUtils}
+import org.apache.spark.sql.execution.datasources.{DataSource, PartitioningUtils}
 import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat
 import org.apache.spark.sql.execution.datasources.json.JsonFileFormat
 import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
@@ -678,12 +675,7 @@ case class DescribeColumnCommand(
     val attribute = {
       val field = catalogTable.schema.find(f => resolver(f.name, column))
       field.getOrElse {
-        if (column.contains(".")) {
-          throw new AnalysisException(
-            s"DESC TABLE COLUMN is not supported for nested column: $column")
-        } else {
-          throw new AnalysisException(s"Column $column does not exist.")
-        }
+        throw new AnalysisException(s"Column $column does not exist.")
       }
     }
 
@@ -700,7 +692,7 @@ case class DescribeColumnCommand(
       // Show column stats only when formatted is specified.
       Row(
         attribute.name,
-        attribute.dataType.simpleString,
+        attribute.dataType.catalogString,
         cs.flatMap(_.min.map(_.toString)).orNull,
         cs.flatMap(_.max.map(_.toString)).orNull,
         cs.map(_.nullCount.toString).orNull,
@@ -711,7 +703,7 @@ case class DescribeColumnCommand(
     } else {
       Row(
         attribute.name,
-        attribute.dataType.simpleString,
+        attribute.dataType.catalogString,
         comment.orNull)
     }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index 3c5e16b935971..785e9b45c8024 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -70,10 +70,6 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
       // Test complex column
       complexData.write.saveAsTable("desc_col_complexTable")
       checkDescColumn("desc_col_complexTable", "s", "struct<key:int,value:string>", None)
-
-      // Describe a nested column
-      val msg = intercept[AnalysisException](sql("DESC desc_col_complexTable s.key")).getMessage
-      assert(msg.contains("DESC TABLE COLUMN is not supported for nested column: s.key"))
     }
   }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
index bc2674c8d6597..2700575691225 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
@@ -236,6 +236,12 @@ class SparkSqlParserSuite extends AnalysisTest {
     assertEqual("DESCRIBE t col",
       DescribeColumnCommand(
         TableIdentifier("t"), "col", isFormatted = false))
+    assertEqual("DESCRIBE t `abc.xyz`",
+      DescribeColumnCommand(
+        TableIdentifier("t"), "abc.xyz", isFormatted = false))
+    // Do not support nested column
+    intercept("DESCRIBE t abc.xyz")
+
     assertEqual("DESCRIBE TABLE t col",
       DescribeColumnCommand(
         TableIdentifier("t"), "col", isFormatted = false))

From 7bc2ea2217a970bbd0dae20c96abfca88f2d692a Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Fri, 7 Jul 2017 15:42:49 +0800
Subject: [PATCH 06/12] throws exception for nested column and add tests in
 .sql and .sql.out

---
 .../spark/sql/catalyst/parser/SqlBase.g4      |   2 +-
 .../catalyst/plans/logical/LogicalPlan.scala  |  26 ++--
 .../spark/sql/execution/SparkSqlParser.scala  |   2 +-
 .../spark/sql/execution/command/tables.scala  |  52 ++++---
 .../inputs/describe-table-column.sql          |  35 +++++
 .../results/describe-table-column.sql.out     | 133 ++++++++++++++++++
 .../org/apache/spark/sql/SQLQuerySuite.scala  |  34 -----
 .../sql/execution/SparkSqlParserSuite.scala   |  18 ++-
 8 files changed, 232 insertions(+), 70 deletions(-)
 create mode 100644 sql/core/src/test/resources/sql-tests/inputs/describe-table-column.sql
 create mode 100644 sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out

diff --git a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
index 557f615d77a35..28b0f6867df4d 100644
--- a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
+++ b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
@@ -262,7 +262,7 @@ describeFuncName
     ;
 
 describeColName
-    : identifier
+    : nameParts+=identifier ('.' nameParts+=identifier)*
     ;
 
 ctes
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
index 8649603b1a9f5..bd38c9e369918 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
@@ -200,10 +200,11 @@ abstract class LogicalPlan
   }
 
   /** Performs attribute resolution given a name and a sequence of possible attributes. */
-  protected def resolve(
+  def resolve(
       nameParts: Seq[String],
       input: Seq[Attribute],
-      resolver: Resolver): Option[NamedExpression] = {
+      resolver: Resolver,
+      resolveNestedFields: Boolean = true): Option[NamedExpression] = {
 
     // A sequence of possible candidate matches.
     // Each candidate is a tuple. The first element is a resolved attribute, followed by a list
@@ -237,14 +238,19 @@ abstract class LogicalPlan
 
       // One match, but we also need to extract the requested nested field.
       case Seq((a, nestedFields)) =>
-        // The foldLeft adds ExtractValues for every remaining parts of the identifier,
-        // and aliased it with the last part of the name.
-        // For example, consider "a.b.c", where "a" is resolved to an existing attribute.
-        // Then this will add ExtractValue("c", ExtractValue("b", a)), and alias the final
-        // expression as "c".
-        val fieldExprs = nestedFields.foldLeft(a: Expression)((expr, fieldName) =>
-          ExtractValue(expr, Literal(fieldName), resolver))
-        Some(Alias(fieldExprs, nestedFields.last)())
+        if (resolveNestedFields) {
+          // The foldLeft adds ExtractValues for every remaining parts of the identifier,
+          // and aliased it with the last part of the name.
+          // For example, consider "a.b.c", where "a" is resolved to an existing attribute.
+          // Then this will add ExtractValue("c", ExtractValue("b", a)), and alias the final
+          // expression as "c".
+          val fieldExprs = nestedFields.foldLeft(a: Expression)((expr, fieldName) =>
+            ExtractValue(expr, Literal(fieldName), resolver))
+          Some(Alias(fieldExprs, nestedFields.last)())
+        } else {
+          throw new AnalysisException(
+            s"Nested column is not supported: $name")
+        }
 
       // No matches.
       case Seq() =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 3d90e4f6a154f..5ee88529c8c03 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -326,7 +326,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
       } else {
         DescribeColumnCommand(
           visitTableIdentifier(ctx.tableIdentifier),
-          ctx.describeColName.getText,
+          ctx.describeColName.nameParts.asScala.map(_.getText),
           ctx.FORMATTED != null)
       }
     } else {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 9fae9e7864288..23bb42dd36d25 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -29,7 +29,7 @@ import org.apache.hadoop.fs.Path
 
 import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionException
+import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException, UnresolvedAttribute}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.CatalogTableType._
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
@@ -635,7 +635,7 @@ case class DescribeTableCommand(
  */
 case class DescribeColumnCommand(
     table: TableIdentifier,
-    column: String,
+    colNameParts: Seq[String],
     isFormatted: Boolean)
   extends RunnableCommand {
 
@@ -678,14 +678,17 @@ case class DescribeColumnCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     val resolver = sparkSession.sessionState.conf.resolver
-    val catalogTable = catalog.getTempViewOrPermanentTableMetadata(table)
+    val relation = sparkSession.table(table).queryExecution.analyzed
     val attribute = {
-      val field = catalogTable.schema.find(f => resolver(f.name, column))
+      val field = relation.resolve(
+        colNameParts, relation.output, resolver, resolveNestedFields = false)
       field.getOrElse {
-        throw new AnalysisException(s"Column $column does not exist.")
+        throw new AnalysisException(s"Column ${UnresolvedAttribute(colNameParts).name} does not " +
+          s"exist")
       }
     }
 
+    val catalogTable = catalog.getTempViewOrPermanentTableMetadata(table)
     val colStats = catalogTable.stats.map(_.colStats).getOrElse(Map.empty)
     val cs = colStats.get(attribute.name)
 
@@ -695,26 +698,41 @@ case class DescribeColumnCommand(
       None
     }
 
-    val result = if (isFormatted) {
+    val fieldValues = if (isFormatted) {
       // Show column stats only when formatted is specified.
-      Row(
+      Seq(
         attribute.name,
         attribute.dataType.catalogString,
-        cs.flatMap(_.min.map(_.toString)).orNull,
-        cs.flatMap(_.max.map(_.toString)).orNull,
-        cs.map(_.nullCount.toString).orNull,
-        cs.map(_.distinctCount.toString).orNull,
-        cs.map(_.avgLen.toString).orNull,
-        cs.map(_.maxLen.toString).orNull,
-        comment.orNull)
+        cs.flatMap(_.min.map(_.toString)).getOrElse("NULL"),
+        cs.flatMap(_.max.map(_.toString)).getOrElse("NULL"),
+        cs.map(_.nullCount.toString).getOrElse("NULL"),
+        cs.map(_.distinctCount.toString).getOrElse("NULL"),
+        cs.map(_.avgLen.toString).getOrElse("NULL"),
+        cs.map(_.maxLen.toString).getOrElse("NULL"),
+        comment.getOrElse("NULL"))
     } else {
-      Row(
+      Seq(
         attribute.name,
         attribute.dataType.catalogString,
-        comment.orNull)
+        comment.getOrElse("NULL"))
     }
 
-    Seq(result)
+    Seq(Row(formatColumnInfo(fieldValues)))
+  }
+
+  /** Do alignment for the result for better readability. */
+  private def formatColumnInfo(fieldValues: Seq[String]): String = {
+    assert(output.length == fieldValues.length)
+    val fieldNames = output.map(_.name)
+    val nameBuilder = new StringBuilder()
+    val valueBuilder = new StringBuilder()
+    fieldNames.zip(fieldValues).foreach { case (name, value) =>
+      // This is for alignment.
+      val len = math.max(name.length, value.length) + 1
+      nameBuilder.append(String.format("%-" + len + "s", name)).append("\t")
+      valueBuilder.append(String.format("%-" + len + "s", value)).append("\t")
+    }
+    nameBuilder.toString() + "\n" + valueBuilder.toString()
   }
 }
 
diff --git a/sql/core/src/test/resources/sql-tests/inputs/describe-table-column.sql b/sql/core/src/test/resources/sql-tests/inputs/describe-table-column.sql
new file mode 100644
index 0000000000000..24870def0316e
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/inputs/describe-table-column.sql
@@ -0,0 +1,35 @@
+-- Test temp table
+CREATE TEMPORARY VIEW desc_col_temp_table (key int COMMENT 'column_comment') USING PARQUET;
+
+DESC desc_col_temp_table key;
+
+DESC EXTENDED desc_col_temp_table key;
+
+DESC FORMATTED desc_col_temp_table key;
+
+-- Describe a column with qualified name
+DESC FORMATTED desc_col_temp_table desc_col_temp_table.key;
+
+-- Describe a non-existent column
+DESC desc_col_temp_table key1;
+
+-- Test persistent table
+CREATE TABLE desc_col_table (key int COMMENT 'column_comment') USING PARQUET;
+
+ANALYZE TABLE desc_col_table COMPUTE STATISTICS FOR COLUMNS key;
+
+DESC desc_col_table key;
+
+DESC EXTENDED desc_col_table key;
+
+DESC FORMATTED desc_col_table key;
+
+-- Test complex columns
+CREATE TABLE desc_col_complex_table (`a.b` int, col struct<x:int, y:string>) USING PARQUET;
+
+DESC FORMATTED desc_col_complex_table `a.b`;
+
+DESC FORMATTED desc_col_complex_table col;
+
+-- Describe a nested column
+DESC FORMATTED desc_col_complex_table col.x;
diff --git a/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
new file mode 100644
index 0000000000000..5a305634baedd
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
@@ -0,0 +1,133 @@
+-- Automatically generated by SQLQueryTestSuite
+-- Number of queries: 15
+
+
+-- !query 0
+CREATE TEMPORARY VIEW desc_col_temp_table (key int COMMENT 'column_comment') USING PARQUET
+-- !query 0 schema
+struct<>
+-- !query 0 output
+
+
+
+-- !query 1
+DESC desc_col_temp_table key
+-- !query 1 schema
+struct<col_name:string,data_type:string,comment:string>
+-- !query 1 output
+col_name 	data_type 	comment        	
+key      	int       	column_comment
+
+
+-- !query 2
+DESC EXTENDED desc_col_temp_table key
+-- !query 2 schema
+struct<col_name:string,data_type:string,comment:string>
+-- !query 2 output
+col_name 	data_type 	comment        	
+key      	int       	column_comment
+
+
+-- !query 3
+DESC FORMATTED desc_col_temp_table key
+-- !query 3 schema
+struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+-- !query 3 output
+col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
+key      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	column_comment
+
+
+-- !query 4
+DESC FORMATTED desc_col_temp_table desc_col_temp_table.key
+-- !query 4 schema
+struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+-- !query 4 output
+col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
+key      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	column_comment
+
+
+-- !query 5
+DESC desc_col_temp_table key1
+-- !query 5 schema
+struct<>
+-- !query 5 output
+org.apache.spark.sql.AnalysisException
+Column key1 does not exist;
+
+
+-- !query 6
+CREATE TABLE desc_col_table (key int COMMENT 'column_comment') USING PARQUET
+-- !query 6 schema
+struct<>
+-- !query 6 output
+
+
+
+-- !query 7
+ANALYZE TABLE desc_col_table COMPUTE STATISTICS FOR COLUMNS key
+-- !query 7 schema
+struct<>
+-- !query 7 output
+
+
+
+-- !query 8
+DESC desc_col_table key
+-- !query 8 schema
+struct<col_name:string,data_type:string,comment:string>
+-- !query 8 output
+col_name 	data_type 	comment        	
+key      	int       	column_comment
+
+
+-- !query 9
+DESC EXTENDED desc_col_table key
+-- !query 9 schema
+struct<col_name:string,data_type:string,comment:string>
+-- !query 9 output
+col_name 	data_type 	comment        	
+key      	int       	column_comment
+
+
+-- !query 10
+DESC FORMATTED desc_col_table key
+-- !query 10 schema
+struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+-- !query 10 output
+col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
+key      	int       	NULL 	NULL 	0         	0              	4           	4           	column_comment
+
+
+-- !query 11
+CREATE TABLE desc_col_complex_table (`a.b` int, col struct<x:int, y:string>) USING PARQUET
+-- !query 11 schema
+struct<>
+-- !query 11 output
+
+
+
+-- !query 12
+DESC FORMATTED desc_col_complex_table `a.b`
+-- !query 12 schema
+struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+-- !query 12 output
+col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment 	
+a.b      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	NULL
+
+
+-- !query 13
+DESC FORMATTED desc_col_complex_table col
+-- !query 13 schema
+struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+-- !query 13 output
+col_name 	data_type              	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment 	
+col      	struct<x:int,y:string> 	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	NULL
+
+
+-- !query 14
+DESC FORMATTED desc_col_complex_table col.x
+-- !query 14 schema
+struct<>
+-- !query 14 output
+org.apache.spark.sql.AnalysisException
+Nested column is not supported: col.x;
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
index 785e9b45c8024..68f61cfab6d2f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -39,40 +39,6 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
 
   setupTestData()
 
-  test("describe table column") {
-    def checkDescColumn(
-        table: String,
-        column: String,
-        dataType: String,
-        comment: Option[String]): Unit = {
-      checkAnswer(sql(s"DESC $table $column"), Row(column, dataType, comment.orNull))
-      checkAnswer(sql(s"DESC EXTENDED $table $column"), Row(column, dataType, comment.orNull))
-      checkAnswer(sql(s"DESC FORMATTED $table $column"),
-        Row(column, dataType, null, null, null, null, null, null, comment.orNull))
-    }
-
-    val comment = "foo bar"
-    // Test temp table
-    withTempView("desc_col_tempTable") {
-      sql(s"CREATE TEMPORARY VIEW desc_col_tempTable (key int COMMENT '$comment') USING PARQUET")
-      checkDescColumn("desc_col_tempTable", "key", "int", Some(comment))
-
-      // Describe a non-existent column
-      val msg = intercept[AnalysisException](sql("DESC desc_col_tempTable key1")).getMessage
-      assert(msg.contains("Column key1 does not exist."))
-    }
-
-    withTable("desc_col_persistentTable", "desc_col_complexTable") {
-      // Test persistent table
-      sql(s"CREATE TABLE desc_col_persistentTable (key int COMMENT '$comment') USING PARQUET")
-      checkDescColumn("desc_col_persistentTable", "key", "int", Some(comment))
-
-      // Test complex column
-      complexData.write.saveAsTable("desc_col_complexTable")
-      checkDescColumn("desc_col_complexTable", "s", "struct<key:int,value:string>", None)
-    }
-  }
-
   test("SPARK-8010: promote numeric to string") {
     val df = Seq((1, 1)).toDF("key", "value")
     df.createOrReplaceTempView("src")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
index e377ab2e33d11..e6051edd9b9c6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
@@ -254,22 +254,26 @@ class SparkSqlParserSuite extends AnalysisTest {
   test("describe table column") {
     assertEqual("DESCRIBE t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), "col", isFormatted = false))
+        TableIdentifier("t"), Seq("col"), isFormatted = false))
     assertEqual("DESCRIBE t `abc.xyz`",
       DescribeColumnCommand(
-        TableIdentifier("t"), "abc.xyz", isFormatted = false))
-    // Do not support nested column
-    intercept("DESCRIBE t abc.xyz")
+        TableIdentifier("t"), Seq("abc.xyz"), isFormatted = false))
+    assertEqual("DESCRIBE t abc.xyz",
+      DescribeColumnCommand(
+        TableIdentifier("t"), Seq("abc", "xyz"), isFormatted = false))
+    assertEqual("DESCRIBE t `a.b`.`x.y`",
+      DescribeColumnCommand(
+        TableIdentifier("t"), Seq("a.b", "x.y"), isFormatted = false))
 
     assertEqual("DESCRIBE TABLE t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), "col", isFormatted = false))
+        TableIdentifier("t"), Seq("col"), isFormatted = false))
     assertEqual("DESCRIBE TABLE EXTENDED t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), "col", isFormatted = false))
+        TableIdentifier("t"), Seq("col"), isFormatted = false))
     assertEqual("DESCRIBE TABLE FORMATTED t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), "col", isFormatted = true))
+        TableIdentifier("t"), Seq("col"), isFormatted = true))
 
     intercept("DESCRIBE TABLE t PARTITION (ds='1970-01-01') col",
       "DESC TABLE COLUMN for a specific partition is not supported")

From f81cc849811c3f0d0bc0cda6067e103559e6c3af Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Fri, 7 Jul 2017 16:04:19 +0800
Subject: [PATCH 07/12] remove unnecessray test case

---
 .../spark/sql/StatisticsCollectionSuite.scala | 23 +------------------
 1 file changed, 1 insertion(+), 22 deletions(-)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
index e3537329de1c1..d9392de37a815 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala
@@ -147,7 +147,7 @@ class StatisticsCollectionSuite extends StatisticsCollectionTestBase with Shared
       BigInt(4444444444L) -> ("4.1 GB", "4.44E+9"),
       BigInt(5555555555555L) -> ("5.1 TB", "5.56E+12"),
       BigInt(6666666666666666L) -> ("5.9 PB", "6.67E+15"),
-      BigInt(1L << 10) * (1L << 60) -> ("1024.0 EB", "1.18E+21"),
+      BigInt(1L << 10 ) * (1L << 60) -> ("1024.0 EB", "1.18E+21"),
       BigInt(1L << 11) * (1L << 60) -> ("2.36E+21 B", "2.36E+21")
     )
     numbers.foreach { case (input, (expectedSize, expectedRows)) =>
@@ -158,27 +158,6 @@ class StatisticsCollectionSuite extends StatisticsCollectionTestBase with Shared
     }
   }
 
-  test("desc column with stats") {
-    withTable("desc_col_statsTable") {
-      val columns = stats.keys.toSeq
-      val df = data.toDF(columns :+ "carray" : _*)
-      df.write.saveAsTable("desc_col_statsTable")
-      sql(s"ANALYZE TABLE desc_col_statsTable COMPUTE STATISTICS FOR COLUMNS " +
-        s"${columns.mkString(", ")}")
-      stats.zip(df.schema).foreach { case ((colName, columnStat), structField) =>
-        checkAnswer(sql(s"DESC FORMATTED desc_col_statsTable ${structField.name}"),
-          Row(colName, structField.dataType.simpleString,
-            columnStat.min.map(_.toString).orNull,
-            columnStat.max.map(_.toString).orNull,
-            columnStat.nullCount.toString,
-            columnStat.distinctCount.toString,
-            columnStat.avgLen.toString,
-            columnStat.maxLen.toString,
-            structField.getComment().orNull))
-      }
-    }
-  }
-
   test("change stats after truncate command") {
     val table = "change_stats_truncate_table"
     withTable(table) {

From e69f1825f50b3b23d605f2e64fcf8bb8a368117e Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wangzhenhua@huawei.com>
Date: Sat, 8 Jul 2017 14:40:59 +0800
Subject: [PATCH 08/12] keep LogicalPlan unchanged and make extended/formatted
 identical

---
 .../catalyst/plans/logical/LogicalPlan.scala  | 26 ++++++----------
 .../spark/sql/execution/SparkSqlParser.scala  |  5 +--
 .../spark/sql/execution/command/tables.scala  | 31 ++++++++++---------
 .../results/describe-table-column.sql.out     | 14 ++++-----
 .../sql/execution/SparkSqlParserSuite.scala   | 14 ++++-----
 5 files changed, 44 insertions(+), 46 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
index bd38c9e369918..8649603b1a9f5 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
@@ -200,11 +200,10 @@ abstract class LogicalPlan
   }
 
   /** Performs attribute resolution given a name and a sequence of possible attributes. */
-  def resolve(
+  protected def resolve(
       nameParts: Seq[String],
       input: Seq[Attribute],
-      resolver: Resolver,
-      resolveNestedFields: Boolean = true): Option[NamedExpression] = {
+      resolver: Resolver): Option[NamedExpression] = {
 
     // A sequence of possible candidate matches.
     // Each candidate is a tuple. The first element is a resolved attribute, followed by a list
@@ -238,19 +237,14 @@ abstract class LogicalPlan
 
       // One match, but we also need to extract the requested nested field.
       case Seq((a, nestedFields)) =>
-        if (resolveNestedFields) {
-          // The foldLeft adds ExtractValues for every remaining parts of the identifier,
-          // and aliased it with the last part of the name.
-          // For example, consider "a.b.c", where "a" is resolved to an existing attribute.
-          // Then this will add ExtractValue("c", ExtractValue("b", a)), and alias the final
-          // expression as "c".
-          val fieldExprs = nestedFields.foldLeft(a: Expression)((expr, fieldName) =>
-            ExtractValue(expr, Literal(fieldName), resolver))
-          Some(Alias(fieldExprs, nestedFields.last)())
-        } else {
-          throw new AnalysisException(
-            s"Nested column is not supported: $name")
-        }
+        // The foldLeft adds ExtractValues for every remaining parts of the identifier,
+        // and aliased it with the last part of the name.
+        // For example, consider "a.b.c", where "a" is resolved to an existing attribute.
+        // Then this will add ExtractValue("c", ExtractValue("b", a)), and alias the final
+        // expression as "c".
+        val fieldExprs = nestedFields.foldLeft(a: Expression)((expr, fieldName) =>
+          ExtractValue(expr, Literal(fieldName), resolver))
+        Some(Alias(fieldExprs, nestedFields.last)())
 
       // No matches.
       case Seq() =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
index 5ee88529c8c03..40d980a49daa4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala
@@ -320,6 +320,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
    * Create a [[DescribeTableCommand]] logical plan.
    */
   override def visitDescribeTable(ctx: DescribeTableContext): LogicalPlan = withOrigin(ctx) {
+    val isExtended = ctx.EXTENDED != null || ctx.FORMATTED != null
     if (ctx.describeColName != null) {
       if (ctx.partitionSpec != null) {
         throw new ParseException("DESC TABLE COLUMN for a specific partition is not supported", ctx)
@@ -327,7 +328,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
         DescribeColumnCommand(
           visitTableIdentifier(ctx.tableIdentifier),
           ctx.describeColName.nameParts.asScala.map(_.getText),
-          ctx.FORMATTED != null)
+          isExtended)
       }
     } else {
       val partitionSpec = if (ctx.partitionSpec != null) {
@@ -343,7 +344,7 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
       DescribeTableCommand(
         visitTableIdentifier(ctx.tableIdentifier),
         partitionSpec,
-        ctx.EXTENDED != null || ctx.FORMATTED != null)
+        isExtended)
     }
   }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 23bb42dd36d25..46a06c71eaab5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -636,13 +636,13 @@ case class DescribeTableCommand(
 case class DescribeColumnCommand(
     table: TableIdentifier,
     colNameParts: Seq[String],
-    isFormatted: Boolean)
+    isExtended: Boolean)
   extends RunnableCommand {
 
   override val output: Seq[Attribute] = {
     // The displayed names are based on Hive.
     // (Link for the corresponding Hive Jira: https://issues.apache.org/jira/browse/HIVE-7050)
-    if (isFormatted) {
+    if (isExtended) {
       Seq(
         AttributeReference("col_name", StringType, nullable = false,
           new MetadataBuilder().putString("comment", "name of the column").build())(),
@@ -679,30 +679,33 @@ case class DescribeColumnCommand(
     val catalog = sparkSession.sessionState.catalog
     val resolver = sparkSession.sessionState.conf.resolver
     val relation = sparkSession.table(table).queryExecution.analyzed
-    val attribute = {
-      val field = relation.resolve(
-        colNameParts, relation.output, resolver, resolveNestedFields = false)
-      field.getOrElse {
+    val field = {
+      relation.resolve(colNameParts, resolver).getOrElse {
         throw new AnalysisException(s"Column ${UnresolvedAttribute(colNameParts).name} does not " +
           s"exist")
       }
     }
+    if (!field.isInstanceOf[Attribute]) {
+      // If the field is not an attribute after `resolve`, then it's a nested field.
+      throw new AnalysisException(s"DESC TABLE COLUMN command is not supported for nested column:" +
+        s" ${UnresolvedAttribute(colNameParts).name}")
+    }
 
     val catalogTable = catalog.getTempViewOrPermanentTableMetadata(table)
     val colStats = catalogTable.stats.map(_.colStats).getOrElse(Map.empty)
-    val cs = colStats.get(attribute.name)
+    val cs = colStats.get(field.name)
 
-    val comment = if (attribute.metadata.contains("comment")) {
-      Option(attribute.metadata.getString("comment"))
+    val comment = if (field.metadata.contains("comment")) {
+      Option(field.metadata.getString("comment"))
     } else {
       None
     }
 
-    val fieldValues = if (isFormatted) {
+    val fieldValues = if (isExtended) {
       // Show column stats only when formatted is specified.
       Seq(
-        attribute.name,
-        attribute.dataType.catalogString,
+        field.name,
+        field.dataType.catalogString,
         cs.flatMap(_.min.map(_.toString)).getOrElse("NULL"),
         cs.flatMap(_.max.map(_.toString)).getOrElse("NULL"),
         cs.map(_.nullCount.toString).getOrElse("NULL"),
@@ -712,8 +715,8 @@ case class DescribeColumnCommand(
         comment.getOrElse("NULL"))
     } else {
       Seq(
-        attribute.name,
-        attribute.dataType.catalogString,
+        field.name,
+        field.dataType.catalogString,
         comment.getOrElse("NULL"))
     }
 
diff --git a/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
index 5a305634baedd..da8fee9eeb66f 100644
--- a/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
@@ -22,10 +22,10 @@ key      	int       	column_comment
 -- !query 2
 DESC EXTENDED desc_col_temp_table key
 -- !query 2 schema
-struct<col_name:string,data_type:string,comment:string>
+struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
 -- !query 2 output
-col_name 	data_type 	comment        	
-key      	int       	column_comment
+col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
+key      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	column_comment
 
 
 -- !query 3
@@ -83,10 +83,10 @@ key      	int       	column_comment
 -- !query 9
 DESC EXTENDED desc_col_table key
 -- !query 9 schema
-struct<col_name:string,data_type:string,comment:string>
+struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
 -- !query 9 output
-col_name 	data_type 	comment        	
-key      	int       	column_comment
+col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
+key      	int       	NULL 	NULL 	0         	0              	4           	4           	column_comment
 
 
 -- !query 10
@@ -130,4 +130,4 @@ DESC FORMATTED desc_col_complex_table col.x
 struct<>
 -- !query 14 output
 org.apache.spark.sql.AnalysisException
-Nested column is not supported: col.x;
+DESC TABLE COLUMN command is not supported for nested column: col.x;
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
index e6051edd9b9c6..52bf080e0a52f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkSqlParserSuite.scala
@@ -254,26 +254,26 @@ class SparkSqlParserSuite extends AnalysisTest {
   test("describe table column") {
     assertEqual("DESCRIBE t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), Seq("col"), isFormatted = false))
+        TableIdentifier("t"), Seq("col"), isExtended = false))
     assertEqual("DESCRIBE t `abc.xyz`",
       DescribeColumnCommand(
-        TableIdentifier("t"), Seq("abc.xyz"), isFormatted = false))
+        TableIdentifier("t"), Seq("abc.xyz"), isExtended = false))
     assertEqual("DESCRIBE t abc.xyz",
       DescribeColumnCommand(
-        TableIdentifier("t"), Seq("abc", "xyz"), isFormatted = false))
+        TableIdentifier("t"), Seq("abc", "xyz"), isExtended = false))
     assertEqual("DESCRIBE t `a.b`.`x.y`",
       DescribeColumnCommand(
-        TableIdentifier("t"), Seq("a.b", "x.y"), isFormatted = false))
+        TableIdentifier("t"), Seq("a.b", "x.y"), isExtended = false))
 
     assertEqual("DESCRIBE TABLE t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), Seq("col"), isFormatted = false))
+        TableIdentifier("t"), Seq("col"), isExtended = false))
     assertEqual("DESCRIBE TABLE EXTENDED t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), Seq("col"), isFormatted = false))
+        TableIdentifier("t"), Seq("col"), isExtended = true))
     assertEqual("DESCRIBE TABLE FORMATTED t col",
       DescribeColumnCommand(
-        TableIdentifier("t"), Seq("col"), isFormatted = true))
+        TableIdentifier("t"), Seq("col"), isExtended = true))
 
     intercept("DESCRIBE TABLE t PARTITION (ds='1970-01-01') col",
       "DESC TABLE COLUMN for a specific partition is not supported")

From 429b99611800a052354fc17c821c4a9caf13183d Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wangzhenhua@huawei.com>
Date: Sat, 8 Jul 2017 14:45:46 +0800
Subject: [PATCH 09/12] also change comment

---
 .../scala/org/apache/spark/sql/execution/command/tables.scala   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 46a06c71eaab5..f431a14b85ab7 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -702,7 +702,7 @@ case class DescribeColumnCommand(
     }
 
     val fieldValues = if (isExtended) {
-      // Show column stats only when formatted is specified.
+      // Show column stats when extended or formatted is specified.
       Seq(
         field.name,
         field.dataType.catalogString,

From 53e4b386082d501cf882b97685ed3f3ec046aaa8 Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wangzhenhua@huawei.com>
Date: Fri, 8 Sep 2017 16:49:10 +0800
Subject: [PATCH 10/12] new output format

---
 .../spark/sql/execution/command/tables.scala  |  87 ++++-----------
 .../results/describe-table-column.sql.out     | 105 +++++++++++++-----
 .../apache/spark/sql/SQLQueryTestSuite.scala  |  10 +-
 3 files changed, 103 insertions(+), 99 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index f431a14b85ab7..79ef81c937fac 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -640,39 +640,12 @@ case class DescribeColumnCommand(
   extends RunnableCommand {
 
   override val output: Seq[Attribute] = {
-    // The displayed names are based on Hive.
-    // (Link for the corresponding Hive Jira: https://issues.apache.org/jira/browse/HIVE-7050)
-    if (isExtended) {
-      Seq(
-        AttributeReference("col_name", StringType, nullable = false,
-          new MetadataBuilder().putString("comment", "name of the column").build())(),
-        AttributeReference("data_type", StringType, nullable = false,
-          new MetadataBuilder().putString("comment", "data type of the column").build())(),
-        AttributeReference("min", StringType, nullable = true,
-          new MetadataBuilder().putString("comment", "min value of the column").build())(),
-        AttributeReference("max", StringType, nullable = true,
-          new MetadataBuilder().putString("comment", "max value of the column").build())(),
-        AttributeReference("num_nulls", StringType, nullable = true,
-          new MetadataBuilder().putString("comment", "number of nulls of the column").build())(),
-        AttributeReference("distinct_count", StringType, nullable = true,
-          new MetadataBuilder().putString("comment", "distinct count of the column").build())(),
-        AttributeReference("avg_col_len", StringType, nullable = true,
-          new MetadataBuilder().putString("comment",
-            "average length of the values of the column").build())(),
-        AttributeReference("max_col_len", StringType, nullable = true,
-          new MetadataBuilder().putString("comment",
-            "maximum length of the values of the column").build())(),
-        AttributeReference("comment", StringType, nullable = true,
-          new MetadataBuilder().putString("comment", "comment of the column").build())())
-    } else {
-      Seq(
-        AttributeReference("col_name", StringType, nullable = false,
-          new MetadataBuilder().putString("comment", "name of the column").build())(),
-        AttributeReference("data_type", StringType, nullable = false,
-          new MetadataBuilder().putString("comment", "data type of the column").build())(),
-        AttributeReference("comment", StringType, nullable = true,
-          new MetadataBuilder().putString("comment", "comment of the column").build())())
-    }
+    Seq(
+      AttributeReference("info_name", StringType, nullable = false,
+        new MetadataBuilder().putString("comment", "name of the column info").build())(),
+      AttributeReference("info_value", StringType, nullable = false,
+        new MetadataBuilder().putString("comment", "value of the column info").build())()
+    )
   }
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -701,41 +674,21 @@ case class DescribeColumnCommand(
       None
     }
 
-    val fieldValues = if (isExtended) {
-      // Show column stats when extended or formatted is specified.
-      Seq(
-        field.name,
-        field.dataType.catalogString,
-        cs.flatMap(_.min.map(_.toString)).getOrElse("NULL"),
-        cs.flatMap(_.max.map(_.toString)).getOrElse("NULL"),
-        cs.map(_.nullCount.toString).getOrElse("NULL"),
-        cs.map(_.distinctCount.toString).getOrElse("NULL"),
-        cs.map(_.avgLen.toString).getOrElse("NULL"),
-        cs.map(_.maxLen.toString).getOrElse("NULL"),
-        comment.getOrElse("NULL"))
-    } else {
-      Seq(
-        field.name,
-        field.dataType.catalogString,
-        comment.getOrElse("NULL"))
-    }
-
-    Seq(Row(formatColumnInfo(fieldValues)))
-  }
-
-  /** Do alignment for the result for better readability. */
-  private def formatColumnInfo(fieldValues: Seq[String]): String = {
-    assert(output.length == fieldValues.length)
-    val fieldNames = output.map(_.name)
-    val nameBuilder = new StringBuilder()
-    val valueBuilder = new StringBuilder()
-    fieldNames.zip(fieldValues).foreach { case (name, value) =>
-      // This is for alignment.
-      val len = math.max(name.length, value.length) + 1
-      nameBuilder.append(String.format("%-" + len + "s", name)).append("\t")
-      valueBuilder.append(String.format("%-" + len + "s", value)).append("\t")
+    val buffer = ArrayBuffer[Row](
+      Row("col_name", field.name),
+      Row("data_type", field.dataType.catalogString),
+      Row("comment", comment.getOrElse("NULL"))
+    )
+    if (isExtended) {
+      // Show column stats when EXTENDED or FORMATTED is specified.
+      buffer += Row("min", cs.flatMap(_.min.map(_.toString)).getOrElse("NULL"))
+      buffer += Row("max", cs.flatMap(_.max.map(_.toString)).getOrElse("NULL"))
+      buffer += Row("num_nulls", cs.map(_.nullCount.toString).getOrElse("NULL"))
+      buffer += Row("distinct_count", cs.map(_.distinctCount.toString).getOrElse("NULL"))
+      buffer += Row("avg_col_len", cs.map(_.avgLen.toString).getOrElse("NULL"))
+      buffer += Row("max_col_len", cs.map(_.maxLen.toString).getOrElse("NULL"))
     }
-    nameBuilder.toString() + "\n" + valueBuilder.toString()
+    buffer
   }
 }
 
diff --git a/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
index da8fee9eeb66f..f28f0fd7de284 100644
--- a/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
@@ -13,37 +13,59 @@ struct<>
 -- !query 1
 DESC desc_col_temp_table key
 -- !query 1 schema
-struct<col_name:string,data_type:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 1 output
-col_name 	data_type 	comment        	
-key      	int       	column_comment
+col_name	key
+data_type	int
+comment	column_comment
 
 
 -- !query 2
 DESC EXTENDED desc_col_temp_table key
 -- !query 2 schema
-struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 2 output
-col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
-key      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	column_comment
+col_name	key
+data_type	int
+comment	column_comment
+min	NULL
+max	NULL
+num_nulls	NULL
+distinct_count	NULL
+avg_col_len	NULL
+max_col_len	NULL
 
 
 -- !query 3
 DESC FORMATTED desc_col_temp_table key
 -- !query 3 schema
-struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 3 output
-col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
-key      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	column_comment
+col_name	key
+data_type	int
+comment	column_comment
+min	NULL
+max	NULL
+num_nulls	NULL
+distinct_count	NULL
+avg_col_len	NULL
+max_col_len	NULL
 
 
 -- !query 4
 DESC FORMATTED desc_col_temp_table desc_col_temp_table.key
 -- !query 4 schema
-struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 4 output
-col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
-key      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	column_comment
+col_name	key
+data_type	int
+comment	column_comment
+min	NULL
+max	NULL
+num_nulls	NULL
+distinct_count	NULL
+avg_col_len	NULL
+max_col_len	NULL
 
 
 -- !query 5
@@ -74,28 +96,43 @@ struct<>
 -- !query 8
 DESC desc_col_table key
 -- !query 8 schema
-struct<col_name:string,data_type:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 8 output
-col_name 	data_type 	comment        	
-key      	int       	column_comment
+col_name	key
+data_type	int
+comment	column_comment
 
 
 -- !query 9
 DESC EXTENDED desc_col_table key
 -- !query 9 schema
-struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 9 output
-col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
-key      	int       	NULL 	NULL 	0         	0              	4           	4           	column_comment
+col_name	key
+data_type	int
+comment	column_comment
+min	NULL
+max	NULL
+num_nulls	0
+distinct_count	0
+avg_col_len	4
+max_col_len	4
 
 
 -- !query 10
 DESC FORMATTED desc_col_table key
 -- !query 10 schema
-struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 10 output
-col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment        	
-key      	int       	NULL 	NULL 	0         	0              	4           	4           	column_comment
+col_name	key
+data_type	int
+comment	column_comment
+min	NULL
+max	NULL
+num_nulls	0
+distinct_count	0
+avg_col_len	4
+max_col_len	4
 
 
 -- !query 11
@@ -109,19 +146,33 @@ struct<>
 -- !query 12
 DESC FORMATTED desc_col_complex_table `a.b`
 -- !query 12 schema
-struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 12 output
-col_name 	data_type 	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment 	
-a.b      	int       	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	NULL
+col_name	a.b
+data_type	int
+comment	NULL
+min	NULL
+max	NULL
+num_nulls	NULL
+distinct_count	NULL
+avg_col_len	NULL
+max_col_len	NULL
 
 
 -- !query 13
 DESC FORMATTED desc_col_complex_table col
 -- !query 13 schema
-struct<col_name:string,data_type:string,min:string,max:string,num_nulls:string,distinct_count:string,avg_col_len:string,max_col_len:string,comment:string>
+struct<info_name:string,info_value:string>
 -- !query 13 output
-col_name 	data_type              	min  	max  	num_nulls 	distinct_count 	avg_col_len 	max_col_len 	comment 	
-col      	struct<x:int,y:string> 	NULL 	NULL 	NULL      	NULL           	NULL        	NULL        	NULL
+col_name	col
+data_type	struct<x:int,y:string>
+comment	NULL
+min	NULL
+max	NULL
+num_nulls	NULL
+distinct_count	NULL
+avg_col_len	NULL
+max_col_len	NULL
 
 
 -- !query 14
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
index d9130fdcfaea6..349cbe41dc9e7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.planning.PhysicalOperation
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.RuleExecutor
 import org.apache.spark.sql.catalyst.util.{fileToString, stringToFile}
-import org.apache.spark.sql.execution.command.DescribeTableCommand
+import org.apache.spark.sql.execution.command.{DescribeColumnCommand, DescribeTableCommand}
 import org.apache.spark.sql.test.SharedSQLContext
 import org.apache.spark.sql.types.StructType
 
@@ -214,11 +214,11 @@ class SQLQueryTestSuite extends QueryTest with SharedSQLContext {
   /** Executes a query and returns the result as (schema of the output, normalized output). */
   private def getNormalizedResult(session: SparkSession, sql: String): (StructType, Seq[String]) = {
     // Returns true if the plan is supposed to be sorted.
-    def needSort(plan: LogicalPlan): Boolean = plan match {
+    def isSorted(plan: LogicalPlan): Boolean = plan match {
       case _: Join | _: Aggregate | _: Generate | _: Sample | _: Distinct => false
-      case _: DescribeTableCommand => true
+      case _: DescribeTableCommand | _: DescribeColumnCommand => true
       case PhysicalOperation(_, _, Sort(_, true, _)) => true
-      case _ => plan.children.iterator.exists(needSort)
+      case _ => plan.children.iterator.exists(isSorted)
     }
 
     try {
@@ -232,7 +232,7 @@ class SQLQueryTestSuite extends QueryTest with SharedSQLContext {
         .replaceAll("Last Access.*", s"Last Access $notIncludedMsg"))
 
       // If the output is not pre-sorted, sort it.
-      if (needSort(df.queryExecution.analyzed)) (schema, answer) else (schema, answer.sorted)
+      if (isSorted(df.queryExecution.analyzed)) (schema, answer) else (schema, answer.sorted)
 
     } catch {
       case a: AnalysisException =>

From 85cc045f6044c508f345f3749ff0012647652985 Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Sun, 10 Sep 2017 11:10:17 +0800
Subject: [PATCH 11/12] fix comment

---
 .../org/apache/spark/sql/execution/command/tables.scala  | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
index 79ef81c937fac..49c98641f6475 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala
@@ -652,16 +652,17 @@ case class DescribeColumnCommand(
     val catalog = sparkSession.sessionState.catalog
     val resolver = sparkSession.sessionState.conf.resolver
     val relation = sparkSession.table(table).queryExecution.analyzed
+
+    val colName = UnresolvedAttribute(colNameParts).name
     val field = {
       relation.resolve(colNameParts, resolver).getOrElse {
-        throw new AnalysisException(s"Column ${UnresolvedAttribute(colNameParts).name} does not " +
-          s"exist")
+        throw new AnalysisException(s"Column $colName does not exist")
       }
     }
     if (!field.isInstanceOf[Attribute]) {
       // If the field is not an attribute after `resolve`, then it's a nested field.
-      throw new AnalysisException(s"DESC TABLE COLUMN command is not supported for nested column:" +
-        s" ${UnresolvedAttribute(colNameParts).name}")
+      throw new AnalysisException(
+        s"DESC TABLE COLUMN command does not support nested data types: $colName")
     }
 
     val catalogTable = catalog.getTempViewOrPermanentTableMetadata(table)

From 0d49ee91508c908daef672a04768c15a9e5c5dba Mon Sep 17 00:00:00 2001
From: Zhenhua Wang <wzh_zju@163.com>
Date: Sun, 10 Sep 2017 13:11:58 +0800
Subject: [PATCH 12/12] fix test print

---
 .../resources/sql-tests/results/describe-table-column.sql.out   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
index f28f0fd7de284..a51eef7c7e1f6 100644
--- a/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/describe-table-column.sql.out
@@ -181,4 +181,4 @@ DESC FORMATTED desc_col_complex_table col.x
 struct<>
 -- !query 14 output
 org.apache.spark.sql.AnalysisException
-DESC TABLE COLUMN command is not supported for nested column: col.x;
+DESC TABLE COLUMN command does not support nested data types: col.x;