[SPARK-52272][SQL] HiveExternalCatalog alter table loses comment updates

Author: szehon-ho

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

@@ -684,11 +684,23 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
       } else {
         (oldTableDef.schema, oldTableDef.partitionColumnNames)
       }
+
+      // alter column comments
+      val newCommentMap = tableDefinition.schema.map(f => (f.name, f.getComment())).toMap
+      val newSchemaWithComments = newSchema.copy(
+        fields = newSchema.fields.map { f =>
+          newCommentMap.getOrElse(f.name, None) match {
+            case Some(comment) => f.withComment(comment)
+            case _ => f
+          }
+        }
+      )
+
       // // Add old table's owner if we need to restore
       val owner = Option(tableDefinition.owner).filter(_.nonEmpty).getOrElse(oldTableDef.owner)
       val newDef = tableDefinition.copy(
         storage = newStorage,
-        schema = newSchema,
+        schema = newSchemaWithComments,
         partitionColumnNames = partitionColumnNames,
         bucketSpec = oldTableDef.bucketSpec,
         properties = newTableProps,

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

@@ -3392,4 +3392,21 @@ class HiveDDLSuite
       )
     }
   }
+
+  test("SPARK-52272: alter table loses column comments") {
+    val tabName = "tab1"
+    withTable(tabName) {
+      val catalog = spark.sessionState.catalog
+
+      sql(s"CREATE TABLE $tabName (col int COMMENT 'comment1') USING PARQUET")
+      val identifier = TableIdentifier(tabName, Some("default"))
+
+      val table = catalog.getTableMetadata(identifier)
+      val newTable = table.copy(schema =
+        new StructType().add("col", IntegerType, nullable = true, "comment2"))
+      catalog.alterTable(newTable)
+      val fetchedTable = catalog.getTableMetadata(identifier)
+      assert(fetchedTable.schema == newTable.schema)
+    }
+  }
 }