minor fix

Author: scwf

sql/core/src/main/scala/org/apache/spark/sql/json/JSONRelation.scala

@@ -43,8 +43,7 @@ private[sql] case class JSONRelation(
 
   private def baseRDD = sqlContext.sparkContext.textFile(fileName)
 
-  override val schema =
-    userSpecifiedSchema.getOrElse(
+  override val schema = userSpecifiedSchema.getOrElse(
     JsonRDD.inferSchema(
       baseRDD,
       samplingRatio,

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

@@ -100,9 +100,15 @@ private[sql] class DDLParser extends StandardTokenParsers with PackratParsers wi
   protected lazy val pair: Parser[(String, String)] = ident ~ stringLit ^^ { case k ~ v => (k,v) }
 
   protected lazy val column: Parser[StructField] =
-    ident ~  ident ^^ { case name ~ typ =>
+  ( ident ~  ident ^^ { case name ~ typ =>
       StructField(name, metastoreTypes.toDataType(typ))
     }
+  |
+    ident ~ ("decimal" ~ "(" ~> numericLit) ~ ("," ~> numericLit <~ ")") ^^ {
+      case name ~ precision ~ scale =>
+        StructField(name, DecimalType(precision.toInt, scale.toInt))
+    }
+  )
 }
 
 /**
@@ -121,8 +127,8 @@ private[sql] class MetastoreTypes extends RegexParsers {
       "bigint" ^^^ LongType |
       "binary" ^^^ BinaryType |
       "boolean" ^^^ BooleanType |
-      fixedDecimalType |                     // Hive 0.13+ decimal with precision/scale
-      "decimal" ^^^ DecimalType.Unlimited |  // Hive 0.12 decimal with no precision/scale
+      fixedDecimalType |                     // decimal with precision/scale
+      "decimal" ^^^ DecimalType.Unlimited |  // decimal with no precision/scale
       "date" ^^^ DateType |
       "timestamp" ^^^ TimestampType |
       "varchar\\((\\d+)\\)".r ^^^ StringType
@@ -204,8 +210,13 @@ private[sql] case class CreateTableUsing(
     }
     val dataSource =
       clazz.newInstance().asInstanceOf[org.apache.spark.sql.sources.SchemaRelationProvider]
-    val relation = dataSource.createRelation(
-      sqlContext, new CaseInsensitiveMap(options), Some(StructType(tableCols)))
+    val relation = if(tableCols.isEmpty) {
+      dataSource.createRelation(
+        sqlContext, new CaseInsensitiveMap(options))
+    } else {
+      dataSource.createRelation(
+        sqlContext, new CaseInsensitiveMap(options), Some(StructType(tableCols)))
+    }
 
     sqlContext.baseRelationToSchemaRDD(relation).registerTempTable(tableName)
     Seq.empty

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

@@ -41,9 +41,7 @@ trait RelationProvider {
    * Note: the parameters' keywords are case insensitive and this insensitivity is enforced
    * by the Map that is passed to the function.
    */
-  def createRelation(
-      sqlContext: SQLContext,
-      parameters: Map[String, String]): BaseRelation
+  def createRelation(sqlContext: SQLContext, parameters: Map[String, String]): BaseRelation
 }
 
 /**

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

@@ -70,7 +70,7 @@ class QueryTest extends PlanTest {
           """.stripMargin)
     }
 
-    if (prepareAnswer(convertedAnswer) != prepareAnswer(sparkAnswer)) {   // issues here, sparkAnswer may be GenericRow[]
+    if (prepareAnswer(convertedAnswer) != prepareAnswer(sparkAnswer)) {
       fail(s"""
         |Results do not match for query:
         |${rdd.logicalPlan}