[SPARK-16663][SQL] desc table should be consistent between data source and hive serde tables

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

@@ -436,11 +436,12 @@ case class DescribeTableCommand(table: TableIdentifier, isExtended: Boolean, isF
 
   private def describePartitionInfo(table: CatalogTable, buffer: ArrayBuffer[Row]): Unit = {
     if (DDLUtils.isDatasourceTable(table)) {
-      val partCols = DDLUtils.getPartitionColumnsFromTableProperties(table)
-      if (partCols.nonEmpty) {
+      val userSpecifiedSchema = DDLUtils.getSchemaFromTableProperties(table)
+      val partColNames = DDLUtils.getPartitionColumnsFromTableProperties(table)
+      for (schema <- userSpecifiedSchema if partColNames.nonEmpty) {
         append(buffer, "# Partition Information", "", "")
-        append(buffer, s"# ${output.head.name}", "", "")
-        partCols.foreach(col => append(buffer, col, "", ""))
+        append(buffer, s"# ${output.head.name}", output(1).name, output(2).name)
+        describeSchema(StructType(partColNames.map(schema(_))), buffer)
       }
     } else {
       if (table.partitionColumns.nonEmpty) {
@@ -526,8 +527,7 @@ case class DescribeTableCommand(table: TableIdentifier, isExtended: Boolean, isF
 
   private def describeSchema(schema: StructType, buffer: ArrayBuffer[Row]): Unit = {
     schema.foreach { column =>
-      val comment = column.getComment().getOrElse("")
-      append(buffer, column.name, column.dataType.simpleString, comment)
+      append(buffer, column.name, column.dataType.simpleString, column.getComment().orNull)
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/sources/DDLTestSuite.scala

@@ -97,21 +97,21 @@ class DDLTestSuite extends DataSourceTest with SharedSQLContext {
       "describe ddlPeople",
       Seq(
         Row("intType", "int", "test comment test1"),
-        Row("stringType", "string", ""),
-        Row("dateType", "date", ""),
-        Row("timestampType", "timestamp", ""),
-        Row("doubleType", "double", ""),
-        Row("bigintType", "bigint", ""),
-        Row("tinyintType", "tinyint", ""),
-        Row("decimalType", "decimal(10,0)", ""),
-        Row("fixedDecimalType", "decimal(5,1)", ""),
-        Row("binaryType", "binary", ""),
-        Row("booleanType", "boolean", ""),
-        Row("smallIntType", "smallint", ""),
-        Row("floatType", "float", ""),
-        Row("mapType", "map<string,string>", ""),
-        Row("arrayType", "array<string>", ""),
-        Row("structType", "struct<f1:string,f2:int>", "")
+        Row("stringType", "string", null),
+        Row("dateType", "date", null),
+        Row("timestampType", "timestamp", null),
+        Row("doubleType", "double", null),
+        Row("bigintType", "bigint", null),
+        Row("tinyintType", "tinyint", null),
+        Row("decimalType", "decimal(10,0)", null),
+        Row("fixedDecimalType", "decimal(5,1)", null),
+        Row("binaryType", "binary", null),
+        Row("booleanType", "boolean", null),
+        Row("smallIntType", "smallint", null),
+        Row("floatType", "float", null),
+        Row("mapType", "map<string,string>", null),
+        Row("arrayType", "array<string>", null),
+        Row("structType", "struct<f1:string,f2:int>", null)
       ))
 
   test("SPARK-7686 DescribeCommand should have correct physical plan output attributes") {

sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala

@@ -748,7 +748,7 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv
       assert(schema === actualSchema)
 
       // Checks the DESCRIBE output.
-      checkAnswer(sql("DESCRIBE spark6655"), Row("int", "int", "") :: Nil)
+      checkAnswer(sql("DESCRIBE spark6655"), Row("int", "int", null) :: Nil)
     }
   }
 

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

@@ -431,6 +431,22 @@ class HiveDDLSuite
     }
   }
 
+  test("desc table for Hive table - partitioned table") {
+    withTable("tbl") {
+      sql("CREATE TABLE tbl(a int) PARTITIONED BY (b int)")
+
+      assert(sql("DESC tbl").collect().containsSlice(
+        Seq(
+          Row("a", "int", null),
+          Row("b", "int", null),
+          Row("# Partition Information", "", ""),
+          Row("# col_name", "data_type", "comment"),
+          Row("b", "int", null)
+        )
+      ))
+    }
+  }
+
   test("desc table for data source table using Hive Metastore") {
     assume(spark.sparkContext.conf.get(CATALOG_IMPLEMENTATION) == "hive")
     val tabName = "tab1"
@@ -621,7 +637,7 @@ class HiveDDLSuite
 
           val desc = sql("DESC FORMATTED t1").collect().toSeq
 
-          assert(desc.contains(Row("id", "bigint", "")))
+          assert(desc.contains(Row("id", "bigint", null)))
         }
       }
     }
@@ -638,13 +654,13 @@ class HiveDDLSuite
 
       assert(formattedDesc.containsSlice(
         Seq(
-          Row("a", "bigint", ""),
-          Row("b", "bigint", ""),
-          Row("c", "bigint", ""),
-          Row("d", "bigint", ""),
+          Row("a", "bigint", null),
+          Row("b", "bigint", null),
+          Row("c", "bigint", null),
+          Row("d", "bigint", null),
           Row("# Partition Information", "", ""),
-          Row("# col_name", "", ""),
-          Row("d", "", ""),
+          Row("# col_name", "data_type", "comment"),
+          Row("d", "bigint", null),
           Row("", "", ""),
           Row("# Detailed Table Information", "", ""),
           Row("Database:", "default", "")

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

@@ -834,8 +834,8 @@ class HiveQuerySuite extends HiveComparisonTest with BeforeAndAfter {
 
     assertResult(
       Array(
-        Row("a", "int", ""),
-        Row("b", "string", ""))
+        Row("a", "int", null),
+        Row("b", "string", null))
     ) {
       sql("DESCRIBE test_describe_commands2")
         .select('col_name, 'data_type, 'comment)