[SPARK-19784][SPARK-25403][SQL] Refresh the table even table stats is empty

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -401,8 +401,6 @@ class SessionCatalog(
     requireDbExists(db)
     requireTableExists(tableIdentifier)
     externalCatalog.alterTableStats(db, table, newStats)
-    // Invalidate the table relation cache
-    refreshTable(identifier)
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala

@@ -67,6 +67,7 @@ case class AnalyzeColumnCommand(
       colStats = tableMeta.stats.map(_.colStats).getOrElse(Map.empty) ++ newColStats)
 
     sessionState.catalog.alterTableStats(tableIdentWithDB, Some(statistics))
+    sessionState.catalog.refreshTable(tableIdentWithDB)
 
     Seq.empty[Row]
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTableCommand.scala

@@ -48,6 +48,7 @@ case class AnalyzeTableCommand(
     val newStats = CommandUtils.compareAndGetNewStats(tableMeta.stats, newTotalSize, newRowCount)
     if (newStats.isDefined) {
       sessionState.catalog.alterTableStats(tableIdentWithDB, newStats)
+      sessionState.catalog.refreshTable(tableIdentWithDB)
     }
 
     Seq.empty[Row]

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

@@ -463,6 +463,7 @@ case class AlterTableAddPartitionCommand(
         catalog.alterTableStats(table.identifier, None)
       }
     }
+    catalog.refreshTable(table.identifier)
     Seq.empty[Row]
   }
 
@@ -548,6 +549,7 @@ case class AlterTableDropPartitionCommand(
       retainData = retainData)
 
     CommandUtils.updateTableStats(sparkSession, table)
+    catalog.refreshTable(table.identifier)
 
     Seq.empty[Row]
   }
@@ -800,6 +802,7 @@ case class AlterTableSetLocationCommand(
     }
 
     CommandUtils.updateTableStats(sparkSession, table)
+    catalog.refreshTable(table.identifier)
     Seq.empty[Row]
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -489,6 +489,7 @@ case class TruncateTableCommand(
       val newStats = CatalogStatistics(sizeInBytes = 0, rowCount = Some(0))
       catalog.alterTableStats(tableName, Some(newStats))
     }
+    catalog.refreshTable(table.identifier)
     Seq.empty[Row]
   }
 }

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

@@ -189,6 +189,7 @@ case class InsertIntoHadoopFsRelationCommand(
       sparkSession.catalog.refreshByPath(outputPath.toString)
 
       if (catalogTable.nonEmpty) {
+        sparkSession.sessionState.catalog.refreshTable(catalogTable.get.identifier)
         CommandUtils.updateTableStats(sparkSession, catalogTable.get)
       }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -26,7 +26,7 @@ import org.scalatest.BeforeAndAfterEach
 
 import org.apache.spark.internal.config
 import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
-import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, NoSuchPartitionException, NoSuchTableException, TempTableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
@@ -167,6 +167,37 @@ class InMemoryCatalogedDDLSuite extends DDLSuite with SharedSQLContext with Befo
       assert(e.message.contains("It doesn't match the specified format"))
     }
   }
+
+  test("SPARK-25403 refresh the table after inserting data") {
+    withTable("t") {
+      val catalog = spark.sessionState.catalog
+      val table = QualifiedTableName(catalog.getCurrentDatabase, "t")
+      sql("CREATE TABLE t (a INT) USING parquet")
+      sql("INSERT INTO TABLE t VALUES (1)")
+      assert(catalog.getCachedTable(table) === null, "Table relation should be invalidated.")
+      assert(spark.table("t").count() === 1)
+      assert(catalog.getCachedTable(table) !== null, "Table relation should be cached.")
+    }
+  }
+
+  test("SPARK-19784 refresh the table after altering the table location") {
+    withTable("t") {
+      withTempDir { dir =>
+        val catalog = spark.sessionState.catalog
+        val table = QualifiedTableName(catalog.getCurrentDatabase, "t")
+        val p1 = s"${dir.getCanonicalPath}/p1"
+        val p2 = s"${dir.getCanonicalPath}/p2"
+        sql(s"CREATE TABLE t (a INT) USING parquet LOCATION '$p1'")
+        sql("INSERT INTO TABLE t VALUES (1)")
+        assert(catalog.getCachedTable(table) === null, "Table relation should be invalidated.")
+        spark.range(5).toDF("a").write.parquet(p2)
+        spark.sql(s"ALTER TABLE t SET LOCATION '$p2'")
+        assert(catalog.getCachedTable(table) === null, "Table relation should be invalidated.")
+        assert(spark.table("t").count() === 5)
+        assert(catalog.getCachedTable(table) !== null, "Table relation should be cached.")
+      }
+    }
+  }
 }
 
 abstract class DDLSuite extends QueryTest with SQLTestUtils {

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

@@ -440,7 +440,7 @@ class HiveParquetMetastoreSuite extends ParquetPartitioningTest {
     def checkCached(tableIdentifier: TableIdentifier): Unit = {
       // Converted test_parquet should be cached.
       getCachedDataSourceTable(tableIdentifier) match {
-        case null => fail("Converted test_parquet should be cached in the cache.")
+        case null => fail(s"Converted ${tableIdentifier.table} should be cached in the cache.")
         case LogicalRelation(_: HadoopFsRelation, _, _, _) => // OK
         case other =>
           fail(
@@ -480,7 +480,7 @@ class HiveParquetMetastoreSuite extends ParquetPartitioningTest {
         |INSERT INTO TABLE test_insert_parquet
         |select a, b from jt
       """.stripMargin)
-    checkCached(tableIdentifier)
+    assert(getCachedDataSourceTable(tableIdentifier) === null)
     // Make sure we can read the data.
     checkAnswer(
       sql("select * from test_insert_parquet"),