Follow up of SPARK-16036

Author: yhuai

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

@@ -369,6 +369,8 @@ case class InsertIntoTable(
     if (table.output.isEmpty) {
       None
     } else {
+      // Note: The parser (visitPartitionSpec in AstBuilder) already turns
+      // keys in partition to their lowercase forms.
       val staticPartCols = partition.filter(_._2.isDefined).keySet
       Some(table.output.filterNot(a => staticPartCols.contains(a.name)))
     }

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

@@ -446,7 +446,7 @@ case class DataSource(
           }.getOrElse(Seq.empty[String])
           // TODO: Case sensitivity.
           val sameColumns =
-            existingPartitionColumns.map(_.toLowerCase) == partitionColumns.map(_.toLowerCase)
+            existingPartitionColumns.map(_.toLowerCase()) == partitionColumns.map(_.toLowerCase())
           if (existingPartitionColumns.size > 0 && !sameColumns) {
             throw new AnalysisException(
               s"""Requested partitioning does not match existing partitioning.

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

@@ -67,7 +67,7 @@ private[sql] class ResolveDataSource(sparkSession: SparkSession) extends Rule[Lo
  * table. It also does data type casting and field renaming, to make sure that the columns to be
  * inserted have the correct data type and fields have the correct names.
  */
-private[sql] object PreprocessTableInsertion extends Rule[LogicalPlan] {
+private[sql] case class PreprocessTableInsertion(conf: SQLConf) extends Rule[LogicalPlan] {
   private def preprocess(
       insert: InsertIntoTable,
       tblName: String,
@@ -84,7 +84,13 @@ private[sql] object PreprocessTableInsertion extends Rule[LogicalPlan] {
     if (insert.partition.nonEmpty) {
       // the query's partitioning must match the table's partitioning
       // this is set for queries like: insert into ... partition (one = "a", two = <expr>)
-      if (insert.partition.keySet != partColNames.toSet) {
+      val samePartitionColumns =
+        if (conf.caseSensitiveAnalysis) {
+          insert.partition.keySet == partColNames.toSet
+        } else {
+          insert.partition.keySet.map(_.toLowerCase) == partColNames.map(_.toLowerCase).toSet
+        }
+      if (!samePartitionColumns) {
         throw new AnalysisException(
           s"""
              |Requested partitioning does not match the table $tblName:
@@ -94,7 +100,8 @@ private[sql] object PreprocessTableInsertion extends Rule[LogicalPlan] {
       }
       expectedColumns.map(castAndRenameChildOutput(insert, _)).getOrElse(insert)
     } else {
-      // All partition columns are dynamic because this InsertIntoTable had no partitioning
+      // All partition columns are dynamic because because the InsertIntoTable command does
+      // not explicitly specify partitioning columns.
       expectedColumns.map(castAndRenameChildOutput(insert, _)).getOrElse(insert)
         .copy(partition = partColNames.map(_ -> None).toMap)
     }

sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala

@@ -111,7 +111,7 @@ private[sql] class SessionState(sparkSession: SparkSession) {
   lazy val analyzer: Analyzer = {
     new Analyzer(catalog, conf) {
       override val extendedResolutionRules =
-        PreprocessTableInsertion ::
+        PreprocessTableInsertion(conf) ::
         new FindDataSourceTable(sparkSession) ::
         DataSourceAnalysis ::
         (if (conf.runSQLonFile) new ResolveDataSource(sparkSession) :: Nil else Nil)

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionState.scala

@@ -65,7 +65,7 @@ private[hive] class HiveSessionState(sparkSession: SparkSession)
         catalog.ParquetConversions ::
         catalog.OrcConversions ::
         catalog.CreateTables ::
-        PreprocessTableInsertion ::
+        PreprocessTableInsertion(conf) ::
         DataSourceAnalysis ::
         (if (conf.runSQLonFile) new ResolveDataSource(sparkSession) :: Nil else Nil)
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertIntoHiveTableSuite.scala

@@ -372,4 +372,24 @@ class InsertIntoHiveTableSuite extends QueryTest with TestHiveSingleton with Bef
       assert(!logical.resolved, "Should not resolve: missing partition data")
     }
   }
+
+  testPartitionedTable(
+    "SPARK-16036: better error message when insert into a table with mismatch schema") {
+    tableName =>
+      val e = intercept[AnalysisException] {
+        sql(s"INSERT INTO TABLE $tableName PARTITION(b=1, c=2) SELECT 1, 2, 3")
+      }
+      assert(e.message.contains("the number of columns are different"))
+  }
+
+  testPartitionedTable(
+    "SPARK-16037: INSERT statement should match columns by position") {
+    tableName =>
+      withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
+        sql(s"INSERT INTO TABLE $tableName SELECT 1, 2 AS c, 3 AS b")
+        checkAnswer(sql(s"SELECT a, b, c FROM $tableName"), Row(1, 2, 3))
+        sql(s"INSERT OVERWRITE TABLE $tableName SELECT 1, 2, 3")
+        checkAnswer(sql(s"SELECT a, b, c FROM $tableName"), Row(1, 2, 3))
+      }
+  }
 }

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

@@ -1033,6 +1033,41 @@ class HiveQuerySuite extends HiveComparisonTest with BeforeAndAfter {
     sql("SELECT * FROM boom").queryExecution.analyzed
   }
 
+  test("SPARK-3810: PreprocessTableInsertion static partitioning support") {
+    val analyzedPlan = {
+      loadTestTable("srcpart")
+      sql("DROP TABLE IF EXISTS withparts")
+      sql("CREATE TABLE withparts LIKE srcpart")
+      sql("INSERT INTO TABLE withparts PARTITION(ds='1', hr='2') SELECT key, value FROM src")
+        .queryExecution.analyzed
+      }
+
+    assertResult(1, "Duplicated project detected\n" + analyzedPlan) {
+      analyzedPlan.collect {
+        case _: Project => ()
+      }.size
+    }
+  }
+
+  test("SPARK-3810: PreprocessTableInsertion dynamic partitioning support") {
+    val analyzedPlan = {
+      loadTestTable("srcpart")
+      sql("DROP TABLE IF EXISTS withparts")
+      sql("CREATE TABLE withparts LIKE srcpart")
+      sql("SET hive.exec.dynamic.partition.mode=nonstrict")
+
+      sql("CREATE TABLE IF NOT EXISTS withparts LIKE srcpart")
+      sql("INSERT INTO TABLE withparts PARTITION(ds, hr) SELECT key, value, '1', '2' FROM src")
+        .queryExecution.analyzed
+    }
+
+    assertResult(2, "Duplicated project detected\n" + analyzedPlan) {
+      analyzedPlan.collect {
+        case _: Project => ()
+      }.size
+    }
+  }
+
   test("parse HQL set commands") {
     // Adapted from its SQL counterpart.
     val testKey = "spark.sql.key.usedfortestonly"

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

@@ -1684,36 +1684,4 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
       )
     }
   }
-
-  test("SPARK-16036: better error message when insert into a table with mismatch schema") {
-    withTable("hive_table", "datasource_table") {
-      sql("CREATE TABLE hive_table(a INT) PARTITIONED BY (b INT, c INT)")
-      sql("CREATE TABLE datasource_table(a INT, b INT, c INT) USING parquet PARTITIONED BY (b, c)")
-      val e1 = intercept[AnalysisException] {
-        sql("INSERT INTO TABLE hive_table PARTITION(b=1, c=2) SELECT 1, 2, 3")
-      }
-      assert(e1.message.contains("the number of columns are different"))
-      val e2 = intercept[AnalysisException] {
-        sql("INSERT INTO TABLE datasource_table PARTITION(b=1, c=2) SELECT 1, 2, 3")
-      }
-      assert(e2.message.contains("the number of columns are different"))
-    }
-  }
-
-  test("SPARK-16037: INSERT statement should match columns by position") {
-    withTable("hive_table", "datasource_table") {
-      sql("CREATE TABLE hive_table(a INT) PARTITIONED BY (b INT, c INT)")
-      sql("CREATE TABLE datasource_table(a INT, b INT, c INT) USING parquet PARTITIONED BY (b, c)")
-
-      withSQLConf("hive.exec.dynamic.partition.mode" -> "nonstrict") {
-        sql("INSERT INTO TABLE hive_table SELECT 1, 2 AS c, 3 AS b")
-        checkAnswer(sql("SELECT a, b, c FROM hive_table"), Row(1, 2, 3))
-        sql("INSERT OVERWRITE TABLE hive_table SELECT 1, 2, 3")
-        checkAnswer(sql("SELECT a, b, c FROM hive_table"), Row(1, 2, 3))
-      }
-
-      sql("INSERT INTO TABLE datasource_table SELECT 1, 2 AS c, 3 AS b")
-      checkAnswer(sql("SELECT a, b, c FROM datasource_table"), Row(1, 2, 3))
-    }
-  }
 }