[SPARK-12257] Non partitioned insert into a partitioned Hive table doesn't fail

Author: dilipbiswal

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

@@ -155,6 +155,11 @@ case class InsertIntoHiveTable(
     val partitionColumns = fileSinkConf.getTableInfo.getProperties.getProperty("partition_columns")
     val partitionColumnNames = Option(partitionColumns).map(_.split("/")).orNull
 
+    // Validate that partition values are specified for partition columns.
+    if (partitionColumnNames != null && partitionColumnNames.size > 0 && partitionSpec.size == 0) {
+      throw new SparkException(ErrorMsg.NEED_PARTITION_ERROR.getMsg)
+    }
+
     // Validate partition spec if there exist any dynamic partitions
     if (numDynamicPartitions > 0) {
       // Report error if dynamic partitioning is not enabled

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

@@ -22,8 +22,9 @@ import java.io.File
 import org.apache.hadoop.hive.conf.HiveConf
 import org.scalatest.BeforeAndAfter
 
+import org.apache.spark.SparkException
+import org.apache.spark.sql._
 import org.apache.spark.sql.execution.QueryExecutionException
-import org.apache.spark.sql.{QueryTest, _}
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
@@ -259,4 +260,14 @@ class InsertIntoHiveTableSuite extends QueryTest with TestHiveSingleton with Bef
 
     sql("DROP TABLE table_with_partition")
   }
+
+  test("Insert into partitioned table with no partition values") {
+    val df = Seq((1, 1), (2, 2), (3, 3)).toDF("c1", "C2")
+    df.registerTempTable("nonpartition_tab")
+    sql("CREATE TABLE partitioned_tab (x INT) PARTITIONED BY (y INT)")
+    intercept[SparkException] {
+      sql("INSERT OVERWRITE TABLE partitioned_tab SELECT c1 FROM nonpartition_tab")
+    }
+    sql("DROP TABLE partitioned_tab")
+  }
 }