Refresh metadata cache in the Catalog in CreateMetastoreDataSourceAsSelect.

yhuai · yhuai · commit b9542ef61989 · 2015-02-27T20:10:36.000-08:00
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala
@@ -248,6 +248,8 @@ case class CreateMetastoreDataSourceAsSelect(
         isExternal)
     }
 
+    // Refresh the cache of the table in the catalog.
+    hiveContext.refreshTable(tableName)
     Seq.empty[Row]
   }
 }
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala
@@ -612,4 +612,56 @@ class MetastoreDataSourcesSuite extends QueryTest with BeforeAndAfterEach {
     val actualSchema = table("wide_schema").schema
     assert(schema === actualSchema)
   }
+
+  test("insert into a table") {
+    def createDF(from: Int, to: Int): DataFrame =
+      createDataFrame((from to to).map(i => Tuple2(i, s"str$i"))).toDF("c1", "c2")
+
+    createDF(0, 9).saveAsTable("insertParquet", "parquet")
+    checkAnswer(
+      sql("SELECT p.c1, p.c2 FROM insertParquet p WHERE p.c1 > 5"),
+      (6 to 9).map(i => Row(i, s"str$i")))
+
+    intercept[AnalysisException] {
+      createDF(10, 19).saveAsTable("insertParquet", "parquet")
+    }
+
+    createDF(10, 19).saveAsTable("insertParquet", "parquet", SaveMode.Append)
+    checkAnswer(
+      sql("SELECT p.c1, p.c2 FROM insertParquet p WHERE p.c1 > 5"),
+      (6 to 19).map(i => Row(i, s"str$i")))
+
+    createDF(20, 29).saveAsTable("insertParquet", "parquet", SaveMode.Append)
+    checkAnswer(
+      sql("SELECT p.c1, c2 FROM insertParquet p WHERE p.c1 > 5 AND p.c1 < 25"),
+      (6 to 24).map(i => Row(i, s"str$i")))
+
+    intercept[AnalysisException] {
+      createDF(30, 39).saveAsTable("insertParquet")
+    }
+
+    createDF(30, 39).saveAsTable("insertParquet", SaveMode.Append)
+    checkAnswer(
+      sql("SELECT p.c1, c2 FROM insertParquet p WHERE p.c1 > 5 AND p.c1 < 35"),
+      (6 to 34).map(i => Row(i, s"str$i")))
+
+    createDF(40, 49).insertInto("insertParquet")
+    checkAnswer(
+      sql("SELECT p.c1, c2 FROM insertParquet p WHERE p.c1 > 5 AND p.c1 < 45"),
+      (6 to 44).map(i => Row(i, s"str$i")))
+
+    createDF(50, 59).saveAsTable("insertParquet", SaveMode.Overwrite)
+    checkAnswer(
+      sql("SELECT p.c1, c2 FROM insertParquet p WHERE p.c1 > 51 AND p.c1 < 55"),
+      (52 to 54).map(i => Row(i, s"str$i")))
+    createDF(60, 69).saveAsTable("insertParquet", SaveMode.Ignore)
+    checkAnswer(
+      sql("SELECT p.c1, c2 FROM insertParquet p"),
+      (50 to 59).map(i => Row(i, s"str$i")))
+
+    createDF(70, 79).insertInto("insertParquet", overwrite = true)
+    checkAnswer(
+      sql("SELECT p.c1, c2 FROM insertParquet p"),
+      (70 to 79).map(i => Row(i, s"str$i")))
+  }
 }

Original file line number	Diff line number	Diff line change
`@@ -248,6 +248,8 @@ case class CreateMetastoreDataSourceAsSelect(`
`248`	`248`	`isExternal)`
`249`	`249`	`}`
`250`	`250`
	`251`	`+ // Refresh the cache of the table in the catalog.`
	`252`	`+ hiveContext.refreshTable(tableName)`
`251`	`253`	`Seq.empty[Row]`
`252`	`254`	`}`
`253`	`255`	`}`