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

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.catalog.refreshTable(catalogTable.get.identifier.quotedString)
         CommandUtils.updateTableStats(sparkSession, catalogTable.get)
       }
 

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"),

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -30,12 +30,12 @@ import org.scalatest.BeforeAndAfterEach
 
 import org.apache.spark.SparkException
 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.{NoSuchPartitionException, TableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.execution.command.{DDLSuite, DDLUtils}
 import org.apache.spark.sql.functions._
-import org.apache.spark.sql.hive.HiveExternalCatalog
+import org.apache.spark.sql.hive.{HiveExternalCatalog, HiveUtils}
 import org.apache.spark.sql.hive.HiveUtils.{CONVERT_METASTORE_ORC, CONVERT_METASTORE_PARQUET}
 import org.apache.spark.sql.hive.orc.OrcFileOperator
 import org.apache.spark.sql.hive.test.TestHiveSingleton
@@ -2370,4 +2370,26 @@ class HiveDDLSuite
       ))
     }
   }
+
+  test("Refresh table after insert into table") {
+    withSQLConf(HiveUtils.CONVERT_METASTORE_PARQUET.key -> "false") {
+      Seq("t1", "t2").foreach { tableName =>
+        withTable(tableName) {
+          if (tableName.equals("t1")) {
+            sql(s"CREATE TABLE $tableName (a INT) STORED AS parquet")
+          } else {
+            sql(s"CREATE TABLE $tableName (a INT) USING parquet")
+          }
+
+          sql(s"INSERT INTO TABLE $tableName VALUES (1)")
+
+          val catalog = spark.sessionState.catalog
+          val qualifiedTableName = QualifiedTableName(catalog.getCurrentDatabase, tableName)
+          val cachedRelation = catalog.getCachedTable(qualifiedTableName)
+          // cachedRelation should be null after refresh table.
+          assert(cachedRelation === null)
+        }
+      }
+    }
+  }
 }