overwrite invalidateTable method to clear table fileStatusCache

Author: yangyx

extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTableCatalog.scala

@@ -386,8 +386,7 @@ class HiveTableCatalog(sparkSession: SparkSession)
         case _: NoSuchTableException =>
           throw new NoSuchTableException(ident)
       }
-      HiveFileStatusCache.getOrCreate(sparkSession,
-        catalogName + "." + catalogTable.qualifiedName).invalidateAll()
+      invalidateTable(ident)
       loadTable(ident)
     }
 
@@ -400,9 +399,7 @@ class HiveTableCatalog(sparkSession: SparkSession)
             ident.asTableIdentifier,
             ignoreIfNotExists = true,
             purge = true /* skip HDFS trash */ )
-          if (table.isInstanceOf[HiveTable]) {
-            table.asInstanceOf[HiveTable].fileIndex.refresh()
-          }
+          invalidateTable(ident)
           true
         } else {
           false
@@ -422,12 +419,14 @@ class HiveTableCatalog(sparkSession: SparkSession)
       // Load table to make sure the table exists
       val table = loadTable(oldIdent)
       catalog.renameTable(oldIdent.asTableIdentifier, newIdent.asTableIdentifier)
-      if (table.isInstanceOf[HiveTable]) {
-        table.asInstanceOf[HiveTable].fileIndex.refresh()
-      }
-
+      invalidateTable(oldIdent)
     }
 
+  override def invalidateTable(ident: Identifier): Unit = {
+    val qualifiedName = s"$catalogName.${ident.namespace().mkString(".")}.${ident.name()}"
+    HiveFileStatusCache.getOrCreate(sparkSession, qualifiedName).invalidateAll()
+  }
+
   private def toOptions(properties: Map[String, String]): Map[String, String] = {
     properties.filterKeys(_.startsWith(TableCatalog.OPTION_PREFIX)).map {
       case (key, value) => key.drop(TableCatalog.OPTION_PREFIX.length) -> value

extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/write/HiveBatchWrite.scala

@@ -21,7 +21,6 @@ import scala.util.control.NonFatal
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.hive.conf.HiveConf
-import org.apache.kyuubi.spark.connector.hive.read.HiveFileStatusCache
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.io.FileCommitProtocol
 import org.apache.spark.sql.SparkSession
@@ -33,6 +32,7 @@ import org.apache.spark.sql.execution.datasources.v2.FileBatchWrite
 import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper.toSQLValue
 import org.apache.spark.sql.types.StringType
 import org.apache.kyuubi.spark.connector.hive.{HiveConnectorUtils, HiveTableCatalog, KyuubiHiveConnectorException}
+import org.apache.spark.sql.connector.catalog.Identifier
 
 class HiveBatchWrite(
     sparkSession: SparkSession,
@@ -68,9 +68,9 @@ class HiveBatchWrite(
 
     // un-cache this table.
     hiveTableCatalog.catalog.invalidateCachedTable(table.identifier)
-    // clear fileStatusCache
-    HiveFileStatusCache.getOrCreate(sparkSession,
-      hiveTableCatalog.name() + "." + table.qualifiedName).invalidateAll()
+    hiveTableCatalog.invalidateTable(
+      Identifier.of(Array(table.identifier.database.getOrElse("")), table.identifier.table)
+    )
 
     val catalog = hiveTableCatalog.catalog
     if (sparkSession.sessionState.conf.autoSizeUpdateEnabled) {

extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/HiveCatalogSuite.scala

@@ -39,7 +39,7 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
 import org.apache.kyuubi.spark.connector.hive.HiveTableCatalog.IdentifierHelper
 import org.apache.kyuubi.spark.connector.hive.KyuubiHiveConnectorConf.{READ_CONVERT_METASTORE_ORC, READ_CONVERT_METASTORE_PARQUET}
-import org.apache.kyuubi.spark.connector.hive.read.HiveScan
+import org.apache.kyuubi.spark.connector.hive.read.{HiveFileStatusCache, HiveScan}
 
 class HiveCatalogSuite extends KyuubiHiveTest {
 
@@ -284,16 +284,29 @@ class HiveCatalogSuite extends KyuubiHiveTest {
   }
 
   test("invalidateTable") {
-    val table = catalog.createTable(testIdent, schema, Array.empty[Transform], emptyProps)
-    // Hive v2 don't cache table
-    catalog.invalidateTable(testIdent)
-
-    val loaded = catalog.loadTable(testIdent)
-
-    assert(table.name == loaded.name)
-    assert(table.schema == loaded.schema)
-    assert(table.properties == loaded.properties)
-    catalog.dropTable(testIdent)
+    withSparkSession() { spark =>
+      val table = catalog.createTable(testIdent, schema, Array.empty[Transform], emptyProps)
+      val qualifiedName = s"$catalogName.${testIdent.namespace().mkString(".")}.${testIdent.name()}"
+      val location = table.asInstanceOf[HiveTable].catalogTable.location
+
+      spark.sql(s"select * from $qualifiedName").collect()
+      assert(HiveFileStatusCache.getOrCreate(spark, qualifiedName)
+        .getLeafFiles(new Path(location)).isDefined)
+
+      catalog.invalidateTable(testIdent)
+      assert(HiveFileStatusCache.getOrCreate(spark, qualifiedName)
+        .getLeafFiles(new Path(location)).isEmpty)
+
+      spark.sql(s"select * from $qualifiedName").collect()
+      assert(HiveFileStatusCache.getOrCreate(spark, qualifiedName)
+        .getLeafFiles(new Path(location)).isDefined)
+
+      val loaded = catalog.loadTable(testIdent)
+      assert(table.name == loaded.name)
+      assert(table.schema == loaded.schema)
+      assert(table.properties == loaded.properties)
+      catalog.dropTable(testIdent)
+    }
   }
 
   test("listNamespaces: fail if missing namespace") {