[SPARK-52683][SQL] Support ExternalCatalog alterTableSchema

### What changes were proposed in this pull request?
Add a new ExternalCatalog and SessionCatalog API alterTableSchema that will supersede alterTableDataSchema.

### Why are the changes needed?
Because ExternalCatalog::alterTableDataSchema takes dataSchema only (without partition columns), we lost the context of the partition column.  This will make it impossible for us to support column order where partition column are not at the end.

See #51342 for context

More generally, this is a better intuitive API than alterTableDataSchema, because the caller no longer needs to strip out partition columns.  Also, it is not immediately intuitive that data schema means without partition columns.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Existing test, move test for alterTableDataSchema to the new API

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #51373 from szehon-ho/alter_table_schema.

Authored-by: Szehon Ho <szehon.apache@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: szehon-ho

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

@@ -120,9 +120,21 @@ trait ExternalCatalog {
    * @param db Database that table to alter schema for exists in
    * @param table Name of table to alter schema for
    * @param newDataSchema Updated data schema to be used for the table.
+   * @deprecated since 4.1.0 use `alterTableSchema` instead.
    */
   def alterTableDataSchema(db: String, table: String, newDataSchema: StructType): Unit
 
+  /**
+   * Alter the schema of a table identified by the provided database and table name.
+   *
+   * All partition columns must be preserved.
+   *
+   * @param db Database that table to alter schema for exists in
+   * @param table Name of table to alter schema for
+   * @param newSchema Updated data schema to be used for the table.
+   */
+  def alterTableSchema(db: String, table: String, newSchema: StructType): Unit
+
   /** Alter the statistics of a table. If `stats` is None, then remove all existing statistics. */
   def alterTableStats(db: String, table: String, stats: Option[CatalogStatistics]): Unit
 

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

@@ -125,6 +125,12 @@ class ExternalCatalogWithListener(delegate: ExternalCatalog)
     postToAll(AlterTableEvent(db, table, AlterTableKind.DATASCHEMA))
   }
 
+  override def alterTableSchema(db: String, table: String, newSchema: StructType): Unit = {
+    postToAll(AlterTablePreEvent(db, table, AlterTableKind.SCHEMA))
+    delegate.alterTableSchema(db, table, newSchema)
+    postToAll(AlterTableEvent(db, table, AlterTableKind.SCHEMA))
+  }
+
   override def alterTableStats(
       db: String,
       table: String,

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

@@ -331,6 +331,21 @@ class InMemoryCatalog(
     catalog(db).tables(table).table = origTable.copy(schema = newSchema)
   }
 
+  override def alterTableSchema(
+      db: String,
+      table: String,
+      newSchema: StructType): Unit = synchronized {
+    requireTableExists(db, table)
+    val origTable = catalog(db).tables(table).table
+
+    val partCols = origTable.partitionColumnNames
+    assert(newSchema.map(_.name).takeRight(partCols.length) == partCols,
+      s"Partition columns ${partCols.mkString("[", ", ", "]")} are only supported at the end of " +
+        s"the new schema ${newSchema.catalogString} for now.")
+
+    catalog(db).tables(table).table = origTable.copy(schema = newSchema)
+  }
+
   override def alterTableStats(
       db: String,
       table: String,

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

@@ -484,6 +484,7 @@ class SessionCatalog(
    *
    * @param identifier TableIdentifier
    * @param newDataSchema Updated data schema to be used for the table
+   * @deprecated since 4.1.0 use `alterTableSchema` instead.
    */
   def alterTableDataSchema(
       identifier: TableIdentifier,
@@ -507,6 +508,25 @@ class SessionCatalog(
     externalCatalog.alterTableDataSchema(db, table, newDataSchema)
   }
 
+  /**
+   * Alter the schema of a table identified by the provided table identifier. All partition columns
+   * must be preserved.
+   *
+   * @param identifier TableIdentifier
+   * @param newSchema Updated schema to be used for the table
+   */
+  def alterTableSchema(
+      identifier: TableIdentifier,
+      newSchema: StructType): Unit = {
+    val qualifiedIdent = qualifyIdentifier(identifier)
+    val db = qualifiedIdent.database.get
+    val table = qualifiedIdent.table
+    requireDbExists(db)
+    requireTableExists(qualifiedIdent)
+
+    externalCatalog.alterTableSchema(db, table, newSchema)
+  }
+
   private def columnNameResolved(
       resolver: Resolver,
       schema: StructType,

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

@@ -126,6 +126,7 @@ case class RenameTableEvent(
 object AlterTableKind extends Enumeration {
   val TABLE = "table"
   val DATASCHEMA = "dataSchema"
+  val SCHEMA = "schema"
   val STATS = "stats"
 }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogEventSuite.scala

@@ -128,9 +128,9 @@ class ExternalCatalogEventSuite extends SparkFunSuite {
 
     // ALTER schema
     val newSchema = new StructType().add("id", "long", nullable = false)
-    catalog.alterTableDataSchema("db5", "tbl1", newSchema)
-    checkEvents(AlterTablePreEvent("db5", "tbl1", AlterTableKind.DATASCHEMA) ::
-      AlterTableEvent("db5", "tbl1", AlterTableKind.DATASCHEMA) :: Nil)
+    catalog.alterTableSchema("db5", "tbl1", newSchema)
+    checkEvents(AlterTablePreEvent("db5", "tbl1", AlterTableKind.SCHEMA) ::
+      AlterTableEvent("db5", "tbl1", AlterTableKind.SCHEMA) :: Nil)
 
     // ALTER stats
     catalog.alterTableStats("db5", "tbl1", None)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/ExternalCatalogSuite.scala

@@ -245,12 +245,15 @@ abstract class ExternalCatalogSuite extends SparkFunSuite {
 
   test("alter table schema") {
     val catalog = newBasicCatalog()
-    val newDataSchema = StructType(Seq(
+    val newSchema = StructType(Seq(
       StructField("col1", IntegerType),
-      StructField("new_field_2", StringType)))
-    catalog.alterTableDataSchema("db2", "tbl1", newDataSchema)
+      StructField("new_field_2", StringType),
+      StructField("a", IntegerType),
+      StructField("b", StringType)))
+    catalog.alterTableSchema("db2", "tbl1", newSchema)
     val newTbl1 = catalog.getTable("db2", "tbl1")
-    assert(newTbl1.dataSchema == newDataSchema)
+    assert(newTbl1.dataSchema == StructType(newSchema.take(2)))
+    assert(newTbl1.schema == newSchema)
   }
 
   test("alter table stats") {
@@ -983,6 +986,32 @@ abstract class ExternalCatalogSuite extends SparkFunSuite {
       "db2", "tbl1", Seq(part1.spec), ignoreIfNotExists = false, purge = false, retainData = false)
     assert(fs.exists(partPath))
   }
+
+  test("SPARK-52683: support alterTableSchema partitioned columns") {
+    val catalog = newBasicCatalog()
+
+    val schema = new StructType()
+      .add("a", IntegerType)
+      .add("b", IntegerType)
+      .add("c", StringType)
+    val table = CatalogTable(
+      identifier = TableIdentifier("t", Some("db1")),
+      tableType = CatalogTableType.MANAGED,
+      storage = storageFormat,
+      schema = schema,
+      partitionColumnNames = Seq("c"),
+      provider = Some("hive"))
+    catalog.createTable(table, ignoreIfExists = false)
+
+    val newSchema = new StructType()
+      .add("b", LongType)
+      .add("a", IntegerType)
+      .add("c", StringType)
+    catalog.alterTableSchema("db1", "t", newSchema)
+
+    assert(catalog.getTable("db1", "t").schema == newSchema)
+    assert(catalog.getTable("db1", "t").partitionColumnNames == Seq("c"))
+  }
 }
 
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala

@@ -564,7 +564,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
     }
   }
 
-  test("alter table add columns") {
+  test("alter data schema add columns") {
     withBasicCatalog { sessionCatalog =>
       sessionCatalog.createTable(newTable("t1", "default"), ignoreIfExists = false)
       val oldTab = sessionCatalog.externalCatalog.getTable("default", "t1")
@@ -580,6 +580,22 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
     }
   }
 
+  test("alter schema add columns") {
+    withBasicCatalog { sessionCatalog =>
+      sessionCatalog.createTable(newTable("t1", "default"), ignoreIfExists = false)
+      val oldTab = sessionCatalog.externalCatalog.getTable("default", "t1")
+      val newSchema = StructType(oldTab.dataSchema.fields ++
+        Seq(StructField("c3", IntegerType)) ++ oldTab.partitionSchema)
+
+      sessionCatalog.alterTableSchema(
+        TableIdentifier("t1", Some("default")),
+        newSchema)
+
+      val newTab = sessionCatalog.externalCatalog.getTable("default", "t1")
+      assert(newTab.schema == newSchema)
+    }
+  }
+
   test("alter table drop columns") {
     withBasicCatalog { sessionCatalog =>
       sessionCatalog.createTable(newTable("t1", "default"), ignoreIfExists = false)

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

@@ -247,8 +247,9 @@ case class AlterTableAddColumnsCommand(
     }
     DDLUtils.checkTableColumns(catalogTable, StructType(colsWithProcessedDefaults))
 
-    val existingSchema = CharVarcharUtils.getRawSchema(catalogTable.dataSchema)
-    catalog.alterTableDataSchema(table, StructType(existingSchema ++ colsWithProcessedDefaults))
+    val existingDataSchema = CharVarcharUtils.getRawSchema(catalogTable.dataSchema)
+    catalog.alterTableSchema(table,
+      StructType(existingDataSchema ++ colsWithProcessedDefaults ++ catalogTable.partitionSchema))
     Seq.empty[Row]
   }
 

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

@@ -309,7 +309,7 @@ class V2SessionCatalog(catalog: SessionCatalog)
             collation = collation, storage = storage))
       }
       if (changes.exists(_.isInstanceOf[TableChange.ColumnChange])) {
-        catalog.alterTableDataSchema(ident.asTableIdentifier, schema)
+        catalog.alterTableSchema(ident.asTableIdentifier, schema)
       }
     } catch {
       case _: NoSuchTableException =>