move check into PreWriteCheck

Author: yjshen

sql/core/src/main/scala/org/apache/spark/sql/sources/DataSourceStrategy.scala

@@ -112,12 +112,6 @@ private[sql] object DataSourceStrategy extends Strategy with Logging {
       val mode = if (overwrite) SaveMode.Overwrite else SaveMode.Append
       execution.ExecutedCommand(InsertIntoHadoopFsRelation(t, query, mode)) :: Nil
 
-    case logical.InsertIntoTable(t, _, _, _, _) =>
-      throw new QueryPlanningException(s"""
-         | Attempt to insert into a RDD-based table: ${t.simpleString}, which is immutable.
-         | Save the RDD to a data source and register the data source as a table before insertion.
-       """.stripMargin)
-
     case _ => Nil
   }
 
@@ -399,5 +393,3 @@ private[sql] object DataSourceStrategy extends Strategy with Logging {
     filters.flatMap(translate)
   }
 }
-
-class QueryPlanningException(message: String) extends Exception(message)

sql/core/src/main/scala/org/apache/spark/sql/sources/rules.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.sources
 
+import org.apache.spark.sql.execution.LogicalRDD
 import org.apache.spark.sql.{SaveMode, AnalysisException}
 import org.apache.spark.sql.catalyst.analysis.{EliminateSubQueries, Catalog}
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Alias}
@@ -119,6 +120,10 @@ private[sql] case class PreWriteCheck(catalog: Catalog) extends (LogicalPlan =>
         // The relation in l is not an InsertableRelation.
         failAnalysis(s"$l does not allow insertion.")
 
+      case logical.InsertIntoTable(t, _, _, _, _) =>
+        failAnalysis(
+          s"Attempt to insert into a RDD-based table: ${t.simpleString} which is immutable.")
+
       case CreateTableUsingAsSelect(tableName, _, _, _, SaveMode.Overwrite, _, query) =>
         // When the SaveMode is Overwrite, we need to check if the table is an input table of
         // the query. If so, we will throw an AnalysisException to let users know it is not allowed.

sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala

@@ -17,15 +17,12 @@
 
 package org.apache.spark.sql
 
-import org.apache.spark.sql.sources.QueryPlanningException
-
 import scala.language.postfixOps
 
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.test.{ExamplePointUDT, ExamplePoint}
 
-
 class DataFrameSuite extends QueryTest {
   import org.apache.spark.sql.TestData._
 
@@ -764,24 +761,35 @@ class DataFrameSuite extends QueryTest {
     assert(!f.getMessage.contains("column2"))
   }
 
-  test("SPARK-6941: Better Error Message for inserting into RDD-based Table") {
+  test("SPARK-6941: Better error message for inserting into RDD-based Table") {
     val df = Seq(Tuple1(1)).toDF("col")
-    df.registerTempTable("rdd_base")
+    val insertion = Seq(Tuple1(2)).toDF("col")
 
-    df.write.parquet("tmp_parquet")
+    // pass case: parquet table (HadoopFsRelation)
+    df.write.mode(SaveMode.Overwrite).parquet("tmp_parquet")
     val pdf = ctx.read.parquet("tmp_parquet")
     pdf.registerTempTable("parquet_base")
+    insertion.write.insertInto("parquet_base")
 
-    df.write.json("tmp_json")
+    // pass case: json table (InsertableRelation)
+    df.write.mode(SaveMode.Overwrite).json("tmp_json")
     val jdf = ctx.read.json("tmp_json")
     jdf.registerTempTable("json_base")
+    insertion.write.mode(SaveMode.Overwrite).insertInto("json_base")
 
-    val insertion = Seq(Tuple1(2)).toDF("col")
-    val e = intercept[QueryPlanningException] {
+    // error cases: insert into a RDD
+    df.registerTempTable("rdd_base")
+    val e1 = intercept[AnalysisException] {
       insertion.write.insertInto("rdd_base")
     }
-    assert(e.getMessage.contains("Attempt to insert into a RDD-based table"))
-    insertion.write.insertInto("parquet_base")
-    insertion.write.mode(SaveMode.Overwrite).insertInto("json_base")
+    assert(e1.getMessage.contains("Attempt to insert into a RDD-based table"))
+
+    // error case: insert into a RDD based on data source
+    val indirectDS = pdf.select("col").filter($"col" > 5)
+    indirectDS.registerTempTable("indirect_ds")
+    val e2 = intercept[AnalysisException] {
+      insertion.write.insertInto("indirect_ds")
+    }
+    assert(e2.getMessage.contains("Attempt to insert into a RDD-based table"))
   }
 }