[SPARK-52281][SQL] Change ALTER TABLE ALTER COLUMN TYPE STRING not to apply default collation if original data type was instance of StringType

### What changes were proposed in this pull request?
Changed `ALTER TABLE ALTER COLUMN TYPE STRING` not to apply default collation if original data type was instance of `StringType`.
```
CREATE TABLE T (C1 CHAR/VARCHAR);
ALTER TABLE T DEFAULT COLLATION UTF8_LCASE;
ALTER TABLE T ALTER COLUMN C1 TYPE STRING COLLATE UTF8_LCASE;
-----------------------------------------------------------------------------------
CREATE TABLE T (C1 STRING [COLLATE XYZ])
ALTER TABLE T DEFAULT COLLATION UTF8_LCASE
ALTER TABLE T ALTER COLUMN C1 TYPE STRING // C1 -> STRING [COLLATE XYZ]
```

### Why are the changes needed?
Bug fix.

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

### How was this patch tested?
Tests added to `DefaultCollationTestSuite`.

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

Closes #51001 from ilicmarkodb/fix_alter_colum_with_collation.

Authored-by: ilicmarkodb <marko.ilic@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: ilicmarkodb

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyDefaultCollationToStringType.scala

@@ -22,9 +22,13 @@ import scala.util.control.NonFatal
 import org.apache.spark.sql.catalyst.expressions.{Cast, DefaultStringProducingExpression, Expression, Literal, SubqueryExpression}
 import org.apache.spark.sql.catalyst.plans.logical.{AddColumns, AlterColumns, AlterColumnSpec, AlterViewAs, ColumnDefinition, CreateTable, CreateTempView, CreateView, LogicalPlan, QualifiedColType, ReplaceColumns, ReplaceTable, TableSpec, V2CreateTablePlan}
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.connector.catalog.{SupportsNamespaces, TableCatalog}
+import org.apache.spark.sql.catalyst.trees.CurrentOrigin
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils.CHAR_VARCHAR_TYPE_STRING_METADATA_KEY
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, SupportsNamespaces, Table, TableCatalog}
 import org.apache.spark.sql.connector.catalog.SupportsNamespaces.PROP_COLLATION
-import org.apache.spark.sql.types.{DataType, StringType}
+import org.apache.spark.sql.errors.DataTypeErrors.toSQLId
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.{CharType, DataType, StringType, StructField, VarcharType}
 
 /**
  * Resolves string types in logical plans by assigning them the appropriate collation. The
@@ -35,12 +39,12 @@ import org.apache.spark.sql.types.{DataType, StringType}
  */
 object ApplyDefaultCollationToStringType extends Rule[LogicalPlan] {
   def apply(plan: LogicalPlan): LogicalPlan = {
-    val planWithResolvedDefaultCollation = resolveDefaultCollation(plan)
+    val preprocessedPlan = resolveDefaultCollation(pruneRedundantAlterColumnTypes(plan))
 
-    fetchDefaultCollation(planWithResolvedDefaultCollation) match {
+    fetchDefaultCollation(preprocessedPlan) match {
       case Some(collation) =>
-        transform(planWithResolvedDefaultCollation, StringType(collation))
-      case None => planWithResolvedDefaultCollation
+        transform(preprocessedPlan, StringType(collation))
+      case None => preprocessedPlan
     }
   }
 
@@ -170,22 +174,86 @@ object ApplyDefaultCollationToStringType extends Rule[LogicalPlan] {
       case p if isCreateOrAlterPlan(p) || AnalysisContext.get.collation.isDefined =>
         transformPlan(p, newType)
 
-      case addCols: AddColumns =>
+      case addCols@AddColumns(_: ResolvedTable, _) =>
         addCols.copy(columnsToAdd = replaceColumnTypes(addCols.columnsToAdd, newType))
 
-      case replaceCols: ReplaceColumns =>
+      case replaceCols@ReplaceColumns(_: ResolvedTable, _) =>
         replaceCols.copy(columnsToAdd = replaceColumnTypes(replaceCols.columnsToAdd, newType))
 
-      case a @ AlterColumns(_, specs: Seq[AlterColumnSpec]) =>
+      case a @ AlterColumns(ResolvedTable(_, _, table: Table, _), specs: Seq[AlterColumnSpec]) =>
         val newSpecs = specs.map {
-          case spec if spec.newDataType.isDefined && hasDefaultStringType(spec.newDataType.get) =>
+          case spec if shouldApplyDefaultCollationToAlterColumn(spec, table) =>
             spec.copy(newDataType = Some(replaceDefaultStringType(spec.newDataType.get, newType)))
           case col => col
         }
         a.copy(specs = newSpecs)
     }
   }
 
+  /**
+   * The column type should not be changed if the original column type is [[StringType]] and the new
+   * type is the default [[StringType]] (i.e., [[StringType]] without an explicit collation).
+   *
+   * Query Example:
+   * {{{
+   *   CREATE TABLE t (c1 STRING COLLATE UNICODE)
+   *   ALTER TABLE t ALTER COLUMN c1 TYPE STRING -- c1 will remain STRING COLLATE UNICODE
+   * }}}
+   */
+  private def pruneRedundantAlterColumnTypes(plan: LogicalPlan): LogicalPlan = {
+    plan match {
+      case alterColumns@AlterColumns(
+      ResolvedTable(_, _, table: Table, _), specs: Seq[AlterColumnSpec]) =>
+        val resolvedSpecs = specs.map { spec =>
+          if (spec.newDataType.isDefined && isStringTypeColumn(spec.column, table) &&
+            isDefaultStringType(spec.newDataType.get)) {
+            spec.copy(newDataType = None)
+          } else {
+            spec
+          }
+        }
+        val newAlterColumns = CurrentOrigin.withOrigin(alterColumns.origin) {
+          alterColumns.copy(specs = resolvedSpecs)
+        }
+        newAlterColumns.copyTagsFrom(alterColumns)
+        newAlterColumns
+      case _ =>
+        plan
+    }
+  }
+
+  private def shouldApplyDefaultCollationToAlterColumn(
+      alterColumnSpec: AlterColumnSpec, table: Table): Boolean = {
+    alterColumnSpec.newDataType.isDefined &&
+      // Applies the default collation only if the original column's type is not StringType.
+      !isStringTypeColumn(alterColumnSpec.column, table) &&
+      hasDefaultStringType(alterColumnSpec.newDataType.get)
+  }
+
+  /**
+   * Checks whether the column's [[DataType]] is [[StringType]] in the given table. Throws an error
+   * if the column is not found.
+   */
+  private def isStringTypeColumn(fieldName: FieldName, table: Table): Boolean = {
+    CatalogV2Util.v2ColumnsToStructType(table.columns())
+      .findNestedField(fieldName.name, includeCollections = true, resolver = conf.resolver)
+      .map {
+        case (_, StructField(_, _: CharType, _, _)) =>
+          false
+        case (_, StructField(_, _: VarcharType, _, _)) =>
+          false
+        case (_, StructField(_, _: StringType, _, metadata))
+          if !metadata.contains(CHAR_VARCHAR_TYPE_STRING_METADATA_KEY) =>
+          true
+        case (_, _) =>
+          false
+      }
+      .getOrElse {
+        throw QueryCompilationErrors.unresolvedColumnError(
+          toSQLId(fieldName.name), table.columns().map(_.name))
+      }
+  }
+
   /**
    * Transforms the given plan, by transforming all expressions in its operators to use the given
    * new type instead of the default string type.

sql/core/src/test/scala/org/apache/spark/sql/collation/DefaultCollationTestSuite.scala

@@ -100,8 +100,12 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
       sql(s"ALTER TABLE $testTable ALTER COLUMN c2 TYPE STRING COLLATE UNICODE")
       assertTableColumnCollation(testTable, "c2", "UNICODE")
 
+      // When using ALTER TABLE ALTER COLUMN TYPE, the column should inherit the table's collation
+      // only if it wasn't a string column before. If the column was already a string, and we're
+      // just changing its type to string (without explicit collation) again, keep the original
+      // collation.
       sql(s"ALTER TABLE $testTable ALTER COLUMN c2 TYPE STRING")
-      assertTableColumnCollation(testTable, "c2", "UTF8_BINARY")
+      assertTableColumnCollation(testTable, "c2", "UNICODE")
     }
   }
 
@@ -148,9 +152,26 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
       sql(s"ALTER TABLE $testTable ALTER COLUMN c1 TYPE STRING COLLATE UNICODE_CI")
       assertTableColumnCollation(testTable, "c1", "UNICODE_CI")
 
+      // When using ALTER TABLE ALTER COLUMN TYPE, the column should inherit the table's collation
+      // only if it wasn't a string column before. If the column was already a string, and we're
+      // just changing its type to string (without explicit collation) again, keep the original
+      // collation.
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c1 TYPE STRING")
+      assertTableColumnCollation(testTable, "c1", "UNICODE_CI")
+
       // alter table add columns with explicit collation, check collation for each column
       sql(s"ALTER TABLE $testTable ADD COLUMN c7 STRING COLLATE SR_CI_AI")
       sql(s"ALTER TABLE $testTable ADD COLUMN c8 STRING COLLATE UTF8_BINARY")
+      assertTableColumnCollation(testTable, "c7", "SR_CI_AI")
+      assertTableColumnCollation(testTable, "c8", "UTF8_BINARY")
+
+      // When using ALTER TABLE ALTER COLUMN TYPE, the column should inherit the table's collation
+      // only if it wasn't a string column before. If the column was already a string, and we're
+      // just changing its type to string (without explicit collation) again, keep the original
+      // collation.
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c8 TYPE STRING")
+      assertTableColumnCollation(testTable, "c8", "UTF8_BINARY")
+
       assertTableColumnCollation(testTable, "c1", "UNICODE_CI")
       assertTableColumnCollation(testTable, "c2", "SR")
       assertTableColumnCollation(testTable, "c3", "UTF8_BINARY")
@@ -162,6 +183,24 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
     }
   }
 
+  test("Alter table alter column type with default collation") {
+    // When using ALTER TABLE ALTER COLUMN TYPE, the column should inherit the table's collation
+    // only if it wasn't a string column before. If the column was already a string, and we're
+    // just changing its type to string (without explicit collation) again, keep the original
+    // collation.
+    withTable(testTable) {
+      sql(s"CREATE TABLE $testTable (c1 STRING, c2 STRING COLLATE UTF8_LCASE, c3 STRING)" +
+        s" DEFAULT COLLATION UTF8_LCASE")
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c1 TYPE STRING")
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c2 TYPE STRING")
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c3 TYPE STRING COLLATE UNICODE")
+
+      assertTableColumnCollation(testTable, "c1", "UTF8_LCASE")
+      assertTableColumnCollation(testTable, "c2", "UTF8_LCASE")
+      assertTableColumnCollation(testTable, "c3", "UNICODE")
+    }
+  }
+
   schemaAndObjectCollationPairs.foreach {
     case (schemaDefaultCollation, tableDefaultCollation) =>
       test(
@@ -221,7 +260,7 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
             sql(s"ALTER SCHEMA $testSchema DEFAULT COLLATION $schemaNewCollation")
 
             // Altering schema default collation should not affect existing objects.
-            addAndAlterColumns(tableDefaultCollation = tableDefaultCollation)
+            addAndAlterColumns(c2Collation = "SR_AI", tableDefaultCollation = tableDefaultCollation)
           }
 
           withTable(testTable) {
@@ -413,12 +452,12 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
         sql(s"CREATE TABLE $testTable (c1 STRING, c2 STRING COLLATE SR_AI) " +
           s"$tableDefaultCollationClause")
 
-        addAndAlterColumns(tableDefaultCollation = resolvedDefaultCollation)
+        addAndAlterColumns(c2Collation = "SR_AI", tableDefaultCollation = resolvedDefaultCollation)
       }
     }
   }
 
-  private def addAndAlterColumns(tableDefaultCollation: String): Unit = {
+  private def addAndAlterColumns(c2Collation: String, tableDefaultCollation: String): Unit = {
     // ADD COLUMN
     sql(s"ALTER TABLE $testTable ADD COLUMN c3 STRING")
     sql(s"ALTER TABLE $testTable ADD COLUMN c4 STRING COLLATE SR_AI")
@@ -432,7 +471,7 @@ abstract class DefaultCollationTestSuite extends QueryTest with SharedSparkSessi
     sql(s"ALTER TABLE $testTable ALTER COLUMN c2 TYPE STRING")
     sql(s"ALTER TABLE $testTable ALTER COLUMN c3 TYPE STRING COLLATE UTF8_BINARY")
     assertTableColumnCollation(testTable, "c1", "UNICODE")
-    assertTableColumnCollation(testTable, "c2", tableDefaultCollation)
+    assertTableColumnCollation(testTable, "c2", c2Collation)
     assertTableColumnCollation(testTable, "c3", "UTF8_BINARY")
   }
 }
@@ -806,6 +845,28 @@ class DefaultCollationTestSuiteV2 extends DefaultCollationTestSuite with Datasou
       }
   }
 
+  test("alter char/varchar column to string type") {
+    withTable(testTable) {
+      sql(s"CREATE TABLE $testTable (c1 VARCHAR(10), c2 CHAR(10)) " +
+        s"DEFAULT COLLATION UTF8_LCASE")
+
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c1 TYPE STRING")
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c2 TYPE STRING")
+      assertTableColumnCollation(testTable, "c1", "UTF8_LCASE")
+      assertTableColumnCollation(testTable, "c2", "UTF8_LCASE")
+    }
+
+    withTable(testTable) {
+      sql(s"CREATE TABLE $testTable (c1 VARCHAR(10), c2 CHAR(10)) " +
+        s"DEFAULT COLLATION UTF8_LCASE")
+
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c1 TYPE STRING COLLATE UNICODE")
+      sql(s"ALTER TABLE $testTable ALTER COLUMN c2 TYPE STRING COLLATE UNICODE")
+      assertTableColumnCollation(testTable, "c1", "UNICODE")
+      assertTableColumnCollation(testTable, "c2", "UNICODE")
+    }
+  }
+
   private def testReplaceColumns(
       schemaDefaultCollation: String, tableDefaultCollation: Option[String] = None): Unit = {
     val (tableDefaultCollationClause, resolvedDefaultCollation) =