Skip to content

[SPARK-22178] [SQL] Refresh Persistent Views by REFRESH TABLE Command #19405

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

[SPARK-22178] [SQL] Refresh Persistent Views by REFRESH TABLE Command #19405

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6650fcf
fix
gatorsmile Oct 1, 2017
d677c77
comment update.
gatorsmile Oct 3, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -474,13 +474,20 @@ class CatalogImpl(sparkSession: SparkSession) extends Catalog {
*/
override def refreshTable(tableName: String): Unit = {
val tableIdent = sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
// Temp tables: refresh (or invalidate) any metadata/data cached in the plan recursively.
// Non-temp tables: refresh the metadata cache.
sessionCatalog.refreshTable(tableIdent)
val tableMetadata = sessionCatalog.getTempViewOrPermanentTableMetadata(tableIdent)
val table = sparkSession.table(tableIdent)

if (tableMetadata.tableType == CatalogTableType.VIEW) {
// Temp or persistent views: refresh (or invalidate) any metadata/data cached
// in the plan recursively.
table.queryExecution.analyzed.foreach(_.refresh())
} else {
// Non-temp tables: refresh the metadata cache.
sessionCatalog.refreshTable(tableIdent)
}

// If this table is cached as an InMemoryRelation, drop the original
// cached version and make the new version cached lazily.
val table = sparkSession.table(tableIdent)
if (isCached(table)) {
// Uncache the logicalPlan.
sparkSession.sharedState.cacheManager.uncacheQuery(table, blocking = true)
Expand Down
14 changes: 11 additions & 3 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetadataCacheSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,14 +31,22 @@ import org.apache.spark.sql.test.SQLTestUtils
class HiveMetadataCacheSuite extends QueryTest with SQLTestUtils with TestHiveSingleton {

test("SPARK-16337 temporary view refresh") {
withTempView("view_refresh") {
checkRefreshView(isTemp = true)
}

test("view refresh") {
Copy link
Contributor

@jiangxb1987 jiangxb1987 Oct 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We didn't cover the persistent view case for refresh, that's why the bug happens...

Copy link
Member Author

@gatorsmile gatorsmile Oct 3, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes. We need to ask contributors for adding more test cases when reviewing the PR.

checkRefreshView(isTemp = false)
}

private def checkRefreshView(isTemp: Boolean) {
withView("view_refresh") {
withTable("view_table") {
// Create a Parquet directory
spark.range(start = 0, end = 100, step = 1, numPartitions = 3)
.write.saveAsTable("view_table")

// Read the table in
spark.table("view_table").filter("id > -1").createOrReplaceTempView("view_refresh")
val temp = if (isTemp) "TEMPORARY" else ""
spark.sql(s"CREATE $temp VIEW view_refresh AS SELECT * FROM view_table WHERE id > -1")
assert(sql("select count(*) from view_refresh").first().getLong(0) == 100)

// Delete a file using the Hadoop file system interface since the path returned by
Expand Down