[SPARK-17980] [SQL] Fix refreshByPath for converted Hive tables

sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala

@@ -343,7 +343,8 @@ abstract class Catalog {
 
  /**
  * Invalidate and refresh all the cached data (and the associated metadata) for any dataframe that
- * contains the given data source path.
+ * contains the given data source path. Path matching is by prefix, i.e. "/" would invalidate
+ * everything that is cached.
  *
  * @since 2.0.0
  */

sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala

@@ -185,9 +185,10 @@ class CacheManager extends Logging {
  plan match {
  case lr: LogicalRelation => lr.relation match {
  case hr: HadoopFsRelation =>
+ val prefixToInvalidate = qualifiedPath.toString
  val invalidate = hr.location.rootPaths
- .map(_.makeQualified(fs.getUri, fs.getWorkingDirectory))
- .contains(qualifiedPath)
+ .map(_.makeQualified(fs.getUri, fs.getWorkingDirectory).toString)
+ .exists(_.startsWith(prefixToInvalidate))
  if (invalidate) hr.location.refresh()
  invalidate
  case _ => false

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/TableFileCatalog.scala

@@ -49,13 +49,18 @@ class TableFileCatalog(
 
  private val baseLocation = catalogTable.storage.locationUri
 
+ // Populated on-demand by calls to cachedAllPartitions
+ private var cachedAllPartitions: ListingFileCatalog = null
+
  override def rootPaths: Seq[Path] = baseLocation.map(new Path(_)).toSeq
 
  override def listFiles(filters: Seq[Expression]): Seq[Partition] = {
  filterPartitions(filters).listFiles(Nil)
  }
 
- override def refresh(): Unit = {}
+ override def refresh(): Unit = synchronized {
+ cachedAllPartitions = null
+ }
 
  /**
  * Returns a [[ListingFileCatalog]] for this table restricted to the subset of partitions
@@ -65,7 +70,7 @@ class TableFileCatalog(
  */
  def filterPartitions(filters: Seq[Expression]): ListingFileCatalog = {
  if (filters.isEmpty) {
-cachedAllPartitions
+allPartitions
  } else {
  filterPartitions0(filters)
  }
@@ -90,9 +95,14 @@ class TableFileCatalog(
  }
 
  // Not used in the hot path of queries when metastore partition pruning is enabled
- lazy val cachedAllPartitions: ListingFileCatalog = filterPartitions0(Nil)
+ def allPartitions: ListingFileCatalog = synchronized {
+ if (cachedAllPartitions == null) {
+ cachedAllPartitions = filterPartitions0(Nil)
+ }
+ cachedAllPartitions
+ }
 
- override def inputFiles: Array[String] = cachedAllPartitions.inputFiles
+ override def inputFiles: Array[String] = allPartitions.inputFiles
 }
 
 /**

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -235,7 +235,7 @@ private[hive] class HiveMetastoreCatalog(sparkSession: SparkSession) extends Log
  if (lazyPruningEnabled) {
  catalog
  } else {
- catalog.cachedAllPartitions
+ catalog.allPartitions
  }
  }
  val partitionSchemaColumnNames = partitionSchema.map(_.name.toLowerCase).toSet

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveMetadataCacheSuite.scala

@@ -80,9 +80,13 @@ class HiveMetadataCacheSuite extends QueryTest with SQLTestUtils with TestHiveSi
  val df = spark.sql("select * from test")
  assert(sql("select * from test").count() == 5)
 
+ def deleteRandomFile(): Unit = {
+ val p = new Path(spark.table("test").inputFiles.head)
+ assert(p.getFileSystem(hiveContext.sessionState.newHadoopConf()).delete(p, true))
+ }
+
  // Delete a file, then assert that we tried to read it. This means the table was cached.
- val p = new Path(spark.table("test").inputFiles.head)
- assert(p.getFileSystem(hiveContext.sessionState.newHadoopConf()).delete(p, true))
+ deleteRandomFile()
  val e = intercept[SparkException] {
  sql("select * from test").count()
  }
@@ -91,6 +95,19 @@ class HiveMetadataCacheSuite extends QueryTest with SQLTestUtils with TestHiveSi
  // Test refreshing the cache.
  spark.catalog.refreshTable("test")
  assert(sql("select * from test").count() == 4)
+ assert(spark.table("test").inputFiles.length == 4)
+
+ // Test refresh by path separately since it goes through different code paths than
+ // refreshTable does.
+ deleteRandomFile()
+ spark.catalog.cacheTable("test")
+ spark.catalog.refreshByPath("/some-invalid-path") // no-op
+ val e2 = intercept[SparkException] {
+ sql("select * from test").count()
+ }
+ assert(e2.getMessage.contains("FileNotFoundException"))
+ spark.catalog.refreshByPath(dir.getAbsolutePath)
+ assert(sql("select * from test").count() == 3)
  }
  }
  }