[SQL] Implement Describe Table for SQLContext

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

@@ -20,9 +20,10 @@ package org.apache.spark.sql.execution
 import org.apache.spark.Logging
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.types.StringType
 import org.apache.spark.sql.{SchemaRDD, SQLConf, SQLContext}
 import org.apache.spark.sql.catalyst.errors.TreeNodeException
-import org.apache.spark.sql.catalyst.expressions.{Row, Attribute}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Row, Attribute}
 import org.apache.spark.sql.catalyst.plans.logical
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 
@@ -171,10 +172,16 @@ case class UncacheTableCommand(tableName: String) extends RunnableCommand {
  */
 @DeveloperApi
 case class DescribeCommand(
-    child: SparkPlan,
-    override val output: Seq[Attribute]) extends RunnableCommand {
+    table: LogicalPlan,
+    isExtended: Boolean) extends RunnableCommand {
 
   override def run(sqlContext: SQLContext) = {
-    child.output.map(field => Row(field.name, field.dataType.toString, null))
+    val plan = sqlContext.executePlan(table).analyzed
+    plan.output.map(field => Row(field.name, field.dataType.toString, null))
   }
+
+  override val output: Seq[Attribute] = Seq(
+    AttributeReference("col_name", StringType, nullable = false)(),
+    AttributeReference("data_type", StringType, nullable = false)(),
+    AttributeReference("comment", StringType, nullable = true)())
 }

sql/core/src/main/scala/org/apache/spark/sql/sources/ddl.scala

@@ -17,13 +17,15 @@
 
 package org.apache.spark.sql.sources
 
+import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+
 import scala.language.implicitConversions
 
 import org.apache.spark.Logging
 import org.apache.spark.sql.{SchemaRDD, SQLContext}
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.AbstractSparkSQLParser
-import org.apache.spark.sql.execution.RunnableCommand
+import org.apache.spark.sql.execution.{DescribeCommand, RunnableCommand}
 import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
 
@@ -59,6 +61,8 @@ private[sql] class DDLParser extends AbstractSparkSQLParser with Logging {
   protected val TABLE = Keyword("TABLE")
   protected val USING = Keyword("USING")
   protected val OPTIONS = Keyword("OPTIONS")
+  protected val DESCRIBE = Keyword("DESCRIBE")
+  protected val EXTENDED = Keyword("EXTENDED")
 
   // Data types.
   protected val STRING = Keyword("STRING")
@@ -80,7 +84,7 @@ private[sql] class DDLParser extends AbstractSparkSQLParser with Logging {
 
   protected lazy val ddl: Parser[LogicalPlan] = createTable
 
-  protected def start: Parser[LogicalPlan] = ddl
+  protected def start: Parser[LogicalPlan] = ddl | describe
 
   /**
    * `CREATE [TEMPORARY] TABLE avroTable
@@ -164,6 +168,13 @@ private[sql] class DDLParser extends AbstractSparkSQLParser with Logging {
     mapType |
     structType |
     primitiveType
+
+  private lazy val describe: Parser[LogicalPlan] =
+    DESCRIBE ~> EXTENDED.? ~ rep1sep(ident, ".") ^^ {
+      case isExtended ~ tableIdent =>
+        DescribeCommand((UnresolvedRelation(tableIdent,None)),false)
+    }
+
 }
 
 object ResolvedDataSource {

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -1034,4 +1034,11 @@ class SQLQuerySuite extends QueryTest with BeforeAndAfterAll {
     rdd.registerTempTable("distinctData")
     checkAnswer(sql("SELECT COUNT(DISTINCT key,value) FROM distinctData"), Row(2))
   }
+
+  test("describe table") {
+    checkAnswer(sql("DESCRIBE EXTENDED testData"),Seq(
+    Row("key","IntegerType",null), Row("value","StringType",null)
+    ))
+  }
+
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveQl.scala

@@ -32,6 +32,7 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.plans.logical
 import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.execution.DescribeCommand
 import org.apache.spark.sql.execution.ExplainCommand
 import org.apache.spark.sql.hive.execution.{HiveNativeCommand, DropTable, AnalyzeTable}
 import org.apache.spark.sql.types._
@@ -46,22 +47,6 @@ import scala.collection.JavaConversions._
  */
 private[hive] case object NativePlaceholder extends Command
 
-/**
- * Returned for the "DESCRIBE [EXTENDED] [dbName.]tableName" command.
- * @param table The table to be described.
- * @param isExtended True if "DESCRIBE EXTENDED" is used. Otherwise, false.
- *                   It is effective only when the table is a Hive table.
- */
-case class DescribeCommand(
-    table: LogicalPlan,
-    isExtended: Boolean) extends Command {
-  override def output = Seq(
-    // Column names are based on Hive.
-    AttributeReference("col_name", StringType, nullable = false)(),
-    AttributeReference("data_type", StringType, nullable = false)(),
-    AttributeReference("comment", StringType, nullable = false)())
-}
-
 /** Provides a mapping from HiveQL statements to catalyst logical plans and expression trees. */
 private[hive] object HiveQl {
   protected val nativeCommands = Seq(

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveStrategies.scala

@@ -229,7 +229,7 @@ private[hive] trait HiveStrategies {
             ExecutedCommand(
               DescribeHiveTableCommand(t, describe.output, describe.isExtended)) :: Nil
           case o: LogicalPlan =>
-            ExecutedCommand(RunnableDescribeCommand(planLater(o), describe.output)) :: Nil
+            ExecutedCommand(RunnableDescribeCommand(o, describe.isExtended)) :: Nil
         }
 
       case _ => Nil

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

@@ -23,7 +23,7 @@ import org.scalatest.{BeforeAndAfterAll, FunSuite, GivenWhenThen}
 
 import org.apache.spark.Logging
 import org.apache.spark.sql.execution.{SetCommand, ExplainCommand}
-import org.apache.spark.sql.hive.DescribeCommand
+import org.apache.spark.sql.execution.DescribeCommand
 import org.apache.spark.sql.catalyst.planning.PhysicalOperation
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.util._