[SPARK-16311][SQL] Improve metadata refresh

[petermaxlee]

What changes were proposed in this pull request?

This patch implements the 3 things specified in SPARK-16311:

(1) Append a message to the FileNotFoundException and say that a workaround is to do explicitly metadata refresh.
(2) Make metadata refresh work on temporary tables/views.
(3) Make metadata refresh work on Datasets/DataFrames, by introducing a Dataset.refresh() method.

And one additional small change:
(4) Merge invalidateTable and refreshTable.

How was this patch tested?

Created a new test suite that creates a temporary directory and then deletes a file from it to verify Spark can read the directory once refresh is called.

[petermaxlee]

cc @rxin

[rxin]

cc @cloud-fan / @liancheng

[gatorsmile]

Before, I tried to merge invalidateTable and refreshTable. @yhuai left the following comment:
#13156 (comment)

I think maybe we can keep them separately?

[gatorsmile]

It will remove the cached data. This is different from what JIRA describes. CC @rxin

[petermaxlee]

Other refresh methods also remove cached data, so I thought this is better.

[gatorsmile]

This new API has different behaviors from the refreshTable API and Refresh Table SQL statement. See the following code:

spark/sql/core/src/main/scala/org/apache/spark/sql/internal/CatalogImpl.scala

Lines 349 to 374 in 02a029d

	/**
	* Refresh the cache entry for a table, if any. For Hive metastore table, the metadata
	* is refreshed.
	*
	* @group cachemgmt
	* @since 2.0.0
	*/
	override def refreshTable(tableName: String): Unit = {
	val tableIdent = sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
	sessionCatalog.refreshTable(tableIdent)
	// If this table is cached as an InMemoryRelation, drop the original
	// cached version and make the new version cached lazily.
	val logicalPlan = sparkSession.sessionState.catalog.lookupRelation(tableIdent)
	// Use lookupCachedData directly since RefreshTable also takes databaseName.
	val isCached = sparkSession.sharedState.cacheManager.lookupCachedData(logicalPlan).nonEmpty
	if (isCached) {
	// Create a data frame to represent the table.
	// TODO: Use uncacheTable once it supports database name.
	val df = Dataset.ofRows(sparkSession, logicalPlan)
	// Uncache the logicalPlan.
	sparkSession.sharedState.cacheManager.uncacheQuery(df, blocking = true)
	// Cache it again.
	sparkSession.sharedState.cacheManager.cacheQuery(df, Some(tableIdent.table))
	}
	}

IMO, if we using the word refresh, we have to make them consistent.

[petermaxlee]

ah ic - we can't unpersist.

[cloud-fan]

We can unpersist, but should persist it again immediately.

[liancheng]

Actually we can and should call unpersist, but we should also call persist()/cache() again so that the Dataset will be cached lazily again with correct data when it gets executed next time. I guess that's also what @gatorsmile meant.

[gatorsmile]

Test cases are not enough to cover the metadata refreshing. The current metadata cache is only used for data source tables. We still could convert Hive tables to data source tables. For example, parquet and orc. Thus, we also need to check the behaviors of these cases.

Try to design more test cases for metadata refreshing, including both positive and negative cases.

[petermaxlee]

What do you mean by both positive and negative cases?

[gatorsmile]

For example, I try to refresh the metadata of a DataFrame that has multiple leaf nodes of MetastoreRelation that are already converted to LogicalRelation. I think we expect the metadata stored in cachedDataSourceTables should be invalidated. The positive case is the table is still cached. The negative case is the table is already uncached.

Update: just correct the contents.

[SparkQA]

Test build #61524 has finished for PR 13989 at commit 82f9bec.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[liancheng]

"Refreshes" instead of "Invalidates"?

[petermaxlee]

Would this work? Traverse the logical plan to find whether it references any catalog relation, and if it does, call catalog.refreshTable("...")?

For example

scala> spark.table("test1").queryExecution.optimizedPlan.asInstanceOf[org.apache.spark.sql.execution.datasources.LogicalRelation].metastoreTableIdentifier
res20: Option[org.apache.spark.sql.catalyst.TableIdentifier] = Some(`default`.`test1`)

[liancheng]

One concern of mine is that, analyzed plan, optimized plan, and executed (physical) plan stored in QueryExecution are all lazy vals, which means that they won't be re-optimized/planned accordingly after refreshing metadata of the corresponding logical plan.

Say we constructed a DataFrame df to join a small table A and a large table B. After calling df.write.parquet(...), analyzed, optimized, and executed plans of df are all computed. Since A is small, the planner may decide to broadcast it, and this decision is reflected in the physical plan.

Next, we add a bunch of files into the directory where table A lives to make it super large, then call df.refresh() to refresh the logical plan. Now, if we try to call df.write.parquet(...) again, the query may probably crash since the physical plan is not refreshed and still thinks that A should be broadcasted.

[liancheng]

In general, I think reconstructing a DataFrame/Dataset or using REFRESH TABLE at application level may be a better approach to solve the problem this PR tries to solve. Did I miss any context here?

[rxin]

I think @liancheng has a good point. Why don't we take out Dataset.refresh() for now?

[petermaxlee]

Alright I will do that and submit a new pull request.

Note that I think the data frame refresh is already possible via table refresh, if a data frame references a table, or if some view references a data frame.

[SparkQA]

Test build #3159 has finished for PR 13989 at commit 82f9bec.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.