[SPARK-15667][SQL]Throw exception if columns number of outputs mismatch the inputs

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

@@ -171,6 +171,16 @@ case class InsertIntoHiveTable(
     val partitionColumns = fileSinkConf.getTableInfo.getProperties.getProperty("partition_columns")
     val partitionColumnNames = Option(partitionColumns).map(_.split("/")).getOrElse(Array.empty)
 
+    // All column names in the format of "<column name 1>,<column name 2>,..."
+    val columnsCnt = tableDesc.getProperties.getProperty("columns").split(",").size
+
+    if (columnsCnt + numDynamicPartitions != child.output.size) {
+      throw new SparkException(s"Cannot insert into target table ${tableDesc.getTableName} " +
+        s"because column number are different: target table has $columnsCnt column(s) and " +
+        s"$numDynamicPartitions dynamic partition column(s), but input has ${child.output.size} " +
+        s"column(s).")
+    }
+
     // By this time, the partition map must match the table's partition columns
     if (partitionColumnNames.toSet != partition.keySet) {
       throw new SparkException(

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

@@ -1068,6 +1068,19 @@ class HiveQuerySuite extends HiveComparisonTest with BeforeAndAfter {
     }
   }
 
+  test("SPARK-15667: Check column number matching with static & dynamic partition insert") {
+    intercept[SparkException] {
+      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='a', hr) SELECT key, value FROM src")
+        .queryExecution.sparkPlan
+    }
+  }
+
   test("parse HQL set commands") {
     // Adapted from its SQL counterpart.
     val testKey = "spark.sql.key.usedfortestonly"