address comments

Author: gatorsmile

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

@@ -311,8 +311,6 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
 
   /**
    * Create a [[CreateTableUsing]] or a [[CreateTableUsingAsSelect]] logical plan.
-   *
-   * TODO add bucketing and partitioning.
    */
   override def visitCreateTableUsing(ctx: CreateTableUsingContext): LogicalPlan = withOrigin(ctx) {
     val (table, temp, ifNotExists, external) = visitCreateTableHeader(ctx.createTableHeader)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala

@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Cast, RowOrd
 import org.apache.spark.sql.catalyst.plans.logical
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.command.CreateTableCommand
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.sources.{BaseRelation, InsertableRelation}
 
@@ -206,6 +207,11 @@ private[sql] case class PreWriteCheck(conf: SQLConf, catalog: SessionCatalog)
         // The relation in l is not an InsertableRelation.
         failAnalysis(s"$l does not allow insertion.")
 
+      case c: CreateTableCommand =>
+        // Duplicates are not allowed in partitionBy
+        // Todo: when bucketBy and sortBy are supported, we also need to ban the duplication.
+        checkDuplicates(c.table.partitionColumnNames, "Partition")
+
       case c: CreateTableUsing =>
         // Duplicates are not allowed in partitionBy/bucketBy/sortBy columns.
         checkDuplicates(c.partitionColumns, "Partition")

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

@@ -207,6 +207,15 @@ class HiveDDLSuite
     }
   }
 
+  test("duplicate columns in partitionBy in CREATE TABLE") {
+    withTable("t") {
+      val e = intercept[AnalysisException] {
+        sql("CREATE TABLE boxes (b INT, c INT) PARTITIONED BY (a INT, a INT)")
+      }
+      assert(e.getMessage.contains("Found duplicate column(s) in Partition: `a`"))
+    }
+  }
+
   test("add/drop partitions - external table") {
     val catalog = spark.sessionState.catalog
     withTempDir { tmpDir =>