fix

Author: gatorsmile

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

@@ -937,7 +937,13 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
 
     selectQuery match {
       case Some(q) => CreateTableAsSelectLogicalPlan(tableDesc, q, ifNotExists)
-      case None => CreateTableCommand(tableDesc, ifNotExists)
+      case None =>
+        val partitionColsInTable = partitionCols.map(_.name).toSet.intersect(cols.map(_.name).toSet)
+        if (partitionColsInTable.nonEmpty) {
+          throw new ParseException(s"Column repeated in partitioning columns: " +
+            partitionColsInTable.mkString("[", ",", "]"), ctx)
+        }
+        CreateTableCommand(tableDesc, ifNotExists)
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandSuite.scala

@@ -319,6 +319,12 @@ class DDLCommandSuite extends PlanTest {
     assert(ct.table.storage.locationUri == Some("/something/anything"))
   }
 
+  test("create table - column repeated in partitioning columns") {
+    val query = "CREATE TABLE tab1 (key INT, value STRING) PARTITIONED BY (key INT, hr STRING)"
+    val e = intercept[ParseException] { parser.parsePlan(query) }
+    assert(e.getMessage.contains("Column repeated in partitioning columns: [key]"))
+  }
+
   test("create table using - with partitioned by") {
     val query = "CREATE TABLE my_tab(a INT, b STRING) USING parquet PARTITIONED BY (a)"
     val expected = CreateTableUsing(