[SPARK-274034][SQL] Table Statisics shall be updated automatically if auto update feature is enabled(spark.sql.statistics.size.autoUpdate.enabled =true)

What changes were proposed in this pull request?

For the  table, INSERT OVERWRITE command statistics are automatically computed by default if user set spark.sql.statistics.size.autoUpdate.enabled =true and the statistics shall be recorded
in metadata store, this is not happening currently  because of  validation table.stats.nonEmpty, the statistics were never recorded for the newly created table, this check  doesn't holds good if auto update property feature is enabled by the user.
As part of fix the autoSizeUpdateEnabled has been pulled up as part of separate validation which will ensure if this feature is enabled the system will calculate the size of the table in every insert command
and the same will be recorded in meta-store.

How was this patch tested?
UT is written and manually verified in cluster.

Tested with unit tests + some internal tests on real cluster.

Author: sujith71955

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

@@ -42,16 +42,14 @@ object CommandUtils extends Logging {
 
   /** Change statistics after changing data by commands. */
   def updateTableStats(sparkSession: SparkSession, table: CatalogTable): Unit = {
-    if (table.stats.nonEmpty) {
-      val catalog = sparkSession.sessionState.catalog
-      if (sparkSession.sessionState.conf.autoSizeUpdateEnabled) {
-        val newTable = catalog.getTableMetadata(table.identifier)
-        val newSize = CommandUtils.calculateTotalSize(sparkSession, newTable)
-        val newStats = CatalogStatistics(sizeInBytes = newSize)
-        catalog.alterTableStats(table.identifier, Some(newStats))
-      } else {
-        catalog.alterTableStats(table.identifier, None)
-      }
+    val catalog = sparkSession.sessionState.catalog
+    if (sparkSession.sessionState.conf.autoSizeUpdateEnabled) {
+      val newTable = catalog.getTableMetadata(table.identifier)
+      val newSize = CommandUtils.calculateTotalSize(sparkSession, newTable)
+      val newStats = CatalogStatistics(sizeInBytes = newSize)
+      catalog.alterTableStats(table.identifier, Some(newStats))
+    } else if (table.stats.nonEmpty) {
+      catalog.alterTableStats(table.identifier, None)
     }
   }
 

sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala

@@ -337,6 +337,22 @@ class StatisticsCollectionSuite extends StatisticsCollectionTestBase with Shared
     }
   }
 
+  test("auto gather stats after insert command") {
+    val table = "change_stats_insert_datasource_table"
+    val autoUpdate = true
+      withSQLConf(SQLConf.AUTO_SIZE_UPDATE_ENABLED.key -> autoUpdate.toString) {
+        withTable(table) {
+          sql(s"CREATE TABLE $table (i int, j string) STORED AS PARQUET")
+          // analyze to get initial stats
+          // insert into command
+          sql(s"INSERT INTO TABLE $table SELECT 1, 'abc'")
+            val stats = getCatalogTable(table).stats
+            assert(stats.isDefined)
+            assert(stats.get.sizeInBytes >= 0)
+        }
+    }
+  }
+
   test("invalidation of tableRelationCache after inserts") {
     val table = "invalidate_catalog_cache_table"
     Seq(false, true).foreach { autoUpdate =>