Follow up of SPARK-16034.

Author: yhuai

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

@@ -435,7 +435,7 @@ case class DataSource(
         // If we are appending to a table that already exists, make sure the partitioning matches
         // up.  If we fail to load the table for whatever reason, ignore the check.
         if (mode == SaveMode.Append) {
-          val existingColumns = Try {
+          val existingPartitionColumns = Try {
             resolveRelation()
               .asInstanceOf[HadoopFsRelation]
               .location
@@ -444,13 +444,14 @@ case class DataSource(
               .fieldNames
               .toSeq
           }.getOrElse(Seq.empty[String])
+          // TODO: Case sensitivity.
           val sameColumns =
-            existingColumns.map(_.toLowerCase) == partitionColumns.map(_.toLowerCase)
-          if (existingColumns.size > 0 && !sameColumns) {
+            existingPartitionColumns.map(_.toLowerCase) == partitionColumns.map(_.toLowerCase)
+          if (existingPartitionColumns.size > 0 && !sameColumns) {
             throw new AnalysisException(
               s"""Requested partitioning does not match existing partitioning.
                  |Existing partitioning columns:
-                 |  ${existingColumns.mkString(", ")}
+                 |  ${existingPartitionColumns.mkString(", ")}
                  |Requested partitioning columns:
                  |  ${partitionColumns.mkString(", ")}
                  |""".stripMargin)

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

@@ -1337,8 +1337,24 @@ class DDLSuite extends QueryTest with SharedSQLContext with BeforeAndAfterEach {
       assert(sql("select * from partitionedTable").collect().size == 1)
       // Inserts new data successfully when partition columns are correctly specified in
       // partitionBy(...).
-      df.write.mode("append").partitionBy("a", "b").saveAsTable("partitionedTable")
-      assert(sql("select * from partitionedTable").collect().size == 2)
+      // TODO: Right now, partition columns are always treated in a case-insensitive way.
+      // See the write method in DataSource.scala.
+      Seq((4, 5, 6)).toDF("a", "B", "c")
+        .write
+        .mode("append")
+        .partitionBy("a", "B")
+        .saveAsTable("partitionedTable")
+
+      Seq((7, 8, 9)).toDF("a", "b", "c")
+        .write
+        .mode("append")
+        .partitionBy("a", "b")
+        .saveAsTable("partitionedTable")
+
+      checkAnswer(
+        sql("select a, b, c from partitionedTable"),
+        Row(1, 2, 3) :: Row(4, 5, 6) :: Row(7, 8, 9) :: Nil
+      )
     }
   }
 }