[SPARK-5196][SQL] Support comment in Create Table Field DDL

Support `comment` in create a table field.
__CREATE TEMPORARY TABLE people(name string `comment` "the name of a person")__

Author: OopsOutOfMemory <[email protected]>

Closes #3999 from OopsOutOfMemory/meta_comment and squashes the following commits:

39150d4 [OopsOutOfMemory] add comment and refine test suite

Author: OopsOutOfMemory

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

@@ -27,7 +27,6 @@ import org.apache.spark.sql.execution.RunnableCommand
 import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
 
-
 /**
  * A parser for foreign DDL commands.
  */
@@ -59,6 +58,7 @@ 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 COMMENT = Keyword("COMMENT")
 
   // Data types.
   protected val STRING = Keyword("STRING")
@@ -111,8 +111,13 @@ private[sql] class DDLParser extends AbstractSparkSQLParser with Logging {
   protected lazy val pair: Parser[(String, String)] = ident ~ stringLit ^^ { case k ~ v => (k,v) }
 
   protected lazy val column: Parser[StructField] =
-    ident ~ dataType ^^ { case columnName ~ typ =>
-      StructField(columnName, typ)
+    ident ~ dataType ~ (COMMENT ~> stringLit).?  ^^ { case columnName ~ typ ~ cm =>
+      val meta = cm match {
+        case Some(comment) =>
+          new MetadataBuilder().putString(COMMENT.str.toLowerCase(), comment).build()
+        case None => Metadata.empty
+      }
+      StructField(columnName, typ, true, meta)
     }
 
   protected lazy val primitiveType: Parser[DataType] =

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

@@ -344,4 +344,24 @@ class TableScanSuite extends DataSourceTest {
     }
     assert(schemaNeeded.getMessage.contains("A schema needs to be specified when using"))
   }
+
+  test("SPARK-5196 schema field with comment") {
+    sql(
+      """
+       |CREATE TEMPORARY TABLE student(name string comment "SN", age int comment "SA", grade int)
+       |USING org.apache.spark.sql.sources.AllDataTypesScanSource
+       |OPTIONS (
+       |  from '1',
+       |  to '10'
+       |)
+       """.stripMargin)
+
+       val planned = sql("SELECT * FROM student").queryExecution.executedPlan
+       val comments = planned.schema.fields.map { field =>
+         if (field.metadata.contains("comment")) field.metadata.getString("comment")
+         else "NO_COMMENT"
+       }.mkString(",")
+
+    assert(comments === "SN,SA,NO_COMMENT")
+  }
 }