make match more specific

Author: yjshen

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

@@ -22,7 +22,7 @@ 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}
 import org.apache.spark.sql.catalyst.plans.logical
-import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoTable, LogicalPlan, Project}
+import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.types.DataType
 
@@ -121,8 +121,11 @@ private[sql] case class PreWriteCheck(catalog: Catalog) extends (LogicalPlan =>
         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.")
+        if (!t.isInstanceOf[LeafNode] || t == OneRowRelation || t.isInstanceOf[LocalRelation]) {
+          failAnalysis(s"Inserting into an RDD-based table is not allowed.")
+        } else {
+          // OK
+        }
 
       case CreateTableUsingAsSelect(tableName, _, _, _, SaveMode.Overwrite, _, query) =>
         // When the SaveMode is Overwrite, we need to check if the table is an input table of

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

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql
 
+import org.apache.spark.sql.catalyst.plans.logical.OneRowRelation
+
 import scala.language.postfixOps
 
 import org.apache.spark.sql.functions._
@@ -762,7 +764,7 @@ class DataFrameSuite extends QueryTest {
   }
 
   test("SPARK-6941: Better error message for inserting into RDD-based Table") {
-    val df = Seq(Tuple1(1)).toDF("col")
+    val df = Seq(Tuple1(1)).toDF()
     val insertion = Seq(Tuple1(2)).toDF("col")
 
     // pass case: parquet table (HadoopFsRelation)
@@ -782,14 +784,21 @@ class DataFrameSuite extends QueryTest {
     val e1 = intercept[AnalysisException] {
       insertion.write.insertInto("rdd_base")
     }
-    assert(e1.getMessage.contains("Attempt to insert into a RDD-based table"))
+    assert(e1.getMessage.contains("Inserting into an RDD-based table is not allowed."))
 
     // error case: insert into a RDD based on data source
-    val indirectDS = pdf.select("col").filter($"col" > 5)
+    val indirectDS = pdf.select("_1").filter($"_1" > 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"))
+    assert(e2.getMessage.contains("Inserting into an RDD-based table is not allowed."))
+
+    // error case: insert into a OneRowRelation
+    new DataFrame(ctx, OneRowRelation).registerTempTable("one_row")
+    val e3 = intercept[AnalysisException] {
+      insertion.write.insertInto("one_row")
+    }
+    assert(e3.getMessage.contains("Inserting into an RDD-based table is not allowed."))
   }
 }