address comments.

Author: gatorsmile

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParserUtils.scala

@@ -50,17 +50,6 @@ object ParserUtils {
     }
   }
 
-  /** Check if duplicate exist in a set of column names. */
-  def checkDuplicateNames[T](colNames: Seq[String], ctx: ParserRuleContext): Unit = {
-    if (colNames.length != colNames.distinct.length) {
-      val duplicateColumns = colNames.groupBy(identity).collect {
-        case (x, ys) if ys.length > 1 => "\"" + x + "\""
-      }
-      throw new ParseException(s"Column repeated in partitioning column(s) or duplicate column " +
-        s"name key in the table definition: ${duplicateColumns.mkString("[", ",", "]")}", ctx)
-    }
-  }
-
   /** Get the code that creates the given node. */
   def source(ctx: ParserRuleContext): String = {
     val stream = ctx.getStart.getInputStream

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

@@ -880,14 +880,27 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
     val properties = Option(ctx.tablePropertyList).map(visitPropertyKeyValues).getOrElse(Map.empty)
     val selectQuery = Option(ctx.query).map(plan)
 
+    // Ensuring whether no duplicate name is used in table definition
+    val colNames = cols.map(_.name)
+    if (colNames.length != colNames.distinct.length) {
+      val duplicateColumns = colNames.groupBy(identity).collect {
+        case (x, ys) if ys.length > 1 => "\"" + x + "\""
+      }
+      throw new ParseException(s"Duplicate column name key(s) in the table definition: " +
+        duplicateColumns.mkString("[", ",", "]"), ctx)
+    }
+
+    // Ensuring the existing columns are not used as partition columns
+    val partitionColsInTable = partitionCols.map(_.name).toSet.intersect(colNames.toSet)
+    if (partitionColsInTable.nonEmpty) {
+      throw new ParseException(s"Column repeated in partitioning column(s): " +
+        partitionColsInTable.map("\"" + _ + "\"").mkString("[", ",", "]"), ctx)
+    }
+
     // Note: Hive requires partition columns to be distinct from the schema, so we need
     // to include the partition columns here explicitly
     val schema = cols ++ partitionCols
 
-    // Ensuring whether no duplicate name is used in table definition;
-    // Also ensuring the existing columns are not used as partition columns
-    checkDuplicateNames(colNames = schema.map(_.name), ctx)
-
     // Storage format
     val defaultStorage: CatalogStorageFormat = {
       val defaultStorageType = conf.getConfString("hive.default.fileformat", "textfile")

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

@@ -322,15 +322,13 @@ class DDLCommandSuite extends PlanTest {
   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 column(s) or " +
-      "duplicate column name key in the table definition: [\"key\"]"))
+    assert(e.getMessage.contains("Column repeated in partitioning column(s): [\"key\"]"))
   }
 
   test("create table - duplicate column name key in the table definition") {
     val query = "CREATE TABLE tab1 (key INT, key STRING)"
     val e = intercept[ParseException] { parser.parsePlan(query) }
-    assert(e.getMessage.contains("Column repeated in partitioning column(s) or " +
-      "duplicate column name key in the table definition: [\"key\"]"))
+    assert(e.getMessage.contains("Duplicate column name key(s) in the table definition: [\"key\"]"))
   }
 
   test("create table using - with partitioned by") {