[SPARK-35531][SQL] Update hive table stats without unnecessary convert

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

@@ -721,19 +721,16 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
       table: String,
       stats: Option[CatalogStatistics]): Unit = withClient {
     requireTableExists(db, table)
-    val rawTable = getRawTable(db, table)
-
-    // convert table statistics to properties so that we can persist them through hive client
-    val statsProperties =
+    val rawHiveTable = client.getRawHiveTable(db, table)
+    val oldProps = client.hiveTableProps(rawHiveTable)
+    val newProps =
       if (stats.isDefined) {
-        statsToProperties(stats.get)
+        oldProps ++ statsToProperties(stats.get)
       } else {
-        new mutable.HashMap[String, String]()
+        oldProps
       }
 
-    val oldTableNonStatsProps = rawTable.properties.filterNot(_._1.startsWith(STATISTICS_PREFIX))
-    val updatedTable = rawTable.copy(properties = oldTableNonStatsProps ++ statsProperties)
-    client.alterTable(updatedTable)
+    client.alterTableProps(rawHiveTable, newProps)
   }
 
   override def getTable(db: String, table: String): CatalogTable = withClient {

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala

@@ -113,6 +113,9 @@ private[hive] trait HiveClient {
   /** Creates a table with the given metadata. */
   def createTable(table: CatalogTable, ignoreIfExists: Boolean): Unit
 
+  /** Get hive table properties. */
+  def hiveTableProps(rawHiveTable: RawHiveTable): Map[String, String]
+
   /** Drop the specified table. */
   def dropTable(dbName: String, tableName: String, ignoreIfNotExists: Boolean, purge: Boolean): Unit
 
@@ -127,6 +130,9 @@ private[hive] trait HiveClient {
    */
   def alterTable(dbName: String, tableName: String, table: CatalogTable): Unit
 
+  /** Alter a table properties */
+  def alterTableProps(rawHiveTable: RawHiveTable, newProps: Map[String, String]): Unit
+
   /**
    * Updates the given table with a new data schema and table properties, and keep everything else
    * unchanged.

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -22,7 +22,7 @@ import java.lang.{Iterable => JIterable}
 import java.lang.reflect.InvocationTargetException
 import java.net.URI
 import java.nio.charset.StandardCharsets.UTF_8
-import java.util.{Locale, Map => JMap}
+import java.util.{HashMap => JHashMap, Locale, Map => JMap}
 import java.util.concurrent.TimeUnit._
 
 import scala.collection.JavaConverters._
@@ -62,7 +62,7 @@ import org.apache.spark.sql.connector.catalog.SupportsNamespaces._
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
 import org.apache.spark.sql.execution.QueryExecutionException
 import org.apache.spark.sql.hive.HiveExternalCatalog
-import org.apache.spark.sql.hive.HiveExternalCatalog.DATASOURCE_SCHEMA
+import org.apache.spark.sql.hive.HiveExternalCatalog.{DATASOURCE_SCHEMA}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.util.{CircularBuffer, Utils}
@@ -580,6 +580,11 @@ private[hive] class HiveClientImpl(
       ignoredProperties = ignoredProperties.toMap)
   }
 
+  override def hiveTableProps(rawHiveTable: RawHiveTable): Map[String, String] = {
+    val hiveTable = rawHiveTable.rawTable.asInstanceOf[HiveTable]
+    hiveTable.getParameters.asScala.toMap.filterNot(kv => HiveStatisticsProperties.contains(kv._1))
+  }
+
   override def createTable(table: CatalogTable, ignoreIfExists: Boolean): Unit = withHiveState {
     verifyColumnDataType(table.dataSchema)
     shim.createTable(client, toHiveTable(table, Some(userName)), ignoreIfExists)
@@ -609,6 +614,16 @@ private[hive] class HiveClientImpl(
     shim.alterTable(client, qualifiedTableName, hiveTable)
   }
 
+  override def alterTableProps(
+      rawHiveTable: RawHiveTable,
+      newProps: Map[String, String]): Unit = withHiveState {
+    val hiveTable = rawHiveTable.rawTable.asInstanceOf[HiveTable]
+    val newPropsMap = new JHashMap[String, String]()
+    newPropsMap.putAll(newProps.asJava)
+    hiveTable.getTTable.setParameters(newPropsMap)
+    shim.alterTable(client, s"${hiveTable.getDbName}.${hiveTable.getTableName}", hiveTable)
+  }
+
   override def alterTableDataSchema(
       dbName: String,
       tableName: String,

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

@@ -26,6 +26,7 @@ import org.scalatest.{BeforeAndAfter, PrivateMethodTester}
 
 import org.apache.spark.SparkException
 import org.apache.spark.sql.{QueryTest, _}
+import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.hive.execution.InsertIntoHiveTable
 import org.apache.spark.sql.hive.test.TestHiveSingleton
@@ -894,12 +895,14 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter
 
           sql(insertString.toLowerCase(Locale.ROOT))
           sql(insertString.toUpperCase(Locale.ROOT))
+          spark.sessionState.catalog.alterTableStats(TableIdentifier("test1"), None)
 
           sql(dropString)
           sql(createSpark.toUpperCase(Locale.ROOT))
 
           sql(insertString.toLowerCase(Locale.ROOT))
           sql(insertString.toUpperCase(Locale.ROOT))
+          spark.sessionState.catalog.alterTableStats(TableIdentifier("test1"), None)
         }
       }
     }