Allow altering column type with char/varchar to follow the Apache Spark behavior

Today we can't alter a char/varchar column of a Delta table to a different but compatible char/varchar type. This is too strict and we should follow Apache Spark to allow it.

GitOrigin-RevId: 84e5550457edfe4075dfd130d689302679f82e8e

Author: cloud-fan

core/src/main/scala/org/apache/spark/sql/delta/commands/alterDeltaTableCommands.scala

@@ -34,6 +34,7 @@ import org.apache.spark.sql.catalyst.analysis.{Resolver, UnresolvedAttribute}
 import org.apache.spark.sql.catalyst.catalog.CatalogUtils
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.logical.{IgnoreCachedData, QualifiedColType}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.connector.catalog.TableCatalog
 import org.apache.spark.sql.connector.catalog.TableChange.{After, ColumnPosition, First}
 import org.apache.spark.sql.execution.command.LeafRunnableCommand
@@ -490,14 +491,19 @@ case class AlterTableChangeColumnDeltaCommand(
         throw DeltaErrors.cannotUpdateOtherField(table.name(), o)
     }
 
-    if (SchemaUtils.canChangeDataType(originalField.dataType, newColumn.dataType, resolver,
+    // Analyzer already validates the char/varchar type change of ALTER COLUMN in
+    // `CheckAnalysis.checkAlterTableCommand`. We should normalize char/varchar type to string type
+    // first (original data type is already normalized as we store char/varchar as string type with
+    // special metadata in the Delta log), then apply Delta-specific checks.
+    val newType = CharVarcharUtils.replaceCharVarcharWithString(newColumn.dataType)
+    if (SchemaUtils.canChangeDataType(originalField.dataType, newType, resolver,
         txn.metadata.columnMappingMode, columnPath :+ originalField.name).nonEmpty) {
       throw DeltaErrors.alterTableChangeColumnException(
         s"'${UnresolvedAttribute(columnPath :+ originalField.name).name}' with type " +
           s"'${originalField.dataType}" +
           s" (nullable = ${originalField.nullable})'",
         s"'${UnresolvedAttribute(Seq(newColumn.name)).name}' with type " +
-          s"'${newColumn.dataType}" +
+          s"'$newType" +
           s" (nullable = ${newColumn.nullable})'")
     }
 

core/src/test/scala/org/apache/spark/sql/delta/DeltaAlterTableTests.scala

@@ -29,10 +29,11 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType, StringType, StructType}
+import org.apache.spark.sql.types._
 import org.apache.spark.util.Utils
 
 trait DeltaAlterTableTestBase
@@ -1370,6 +1371,42 @@ trait DeltaAlterTableTests extends DeltaAlterTableTestBase {
       }
     }
   }
+
+  test("CHANGE COLUMN: allow to change change column from char to string type") {
+    withTable("t") {
+      sql("CREATE TABLE t(i STRING, c CHAR(4)) USING delta")
+      sql("ALTER TABLE t CHANGE COLUMN c TYPE STRING")
+      assert(spark.table("t").schema(1).dataType === StringType)
+    }
+  }
+
+  private def checkColType(f: StructField, dt: DataType): Unit = {
+    assert(f.dataType == CharVarcharUtils.replaceCharVarcharWithString(dt))
+    assert(CharVarcharUtils.getRawType(f.metadata).contains(dt))
+  }
+
+  test("CHANGE COLUMN: allow to change column from char(x) to varchar(y) type x <= y") {
+    withTable("t") {
+      sql("CREATE TABLE t(i STRING, c CHAR(4)) USING delta")
+      sql("ALTER TABLE t CHANGE COLUMN c TYPE VARCHAR(4)")
+      checkColType(spark.table("t").schema(1), VarcharType(4))
+    }
+    withTable("t") {
+      sql("CREATE TABLE t(i STRING, c CHAR(4)) USING delta")
+      sql("ALTER TABLE t CHANGE COLUMN c TYPE VARCHAR(5)")
+      checkColType(spark.table("t").schema(1), VarcharType(5))
+    }
+  }
+
+  test("CHANGE COLUMN: allow to change column from varchar(x) to varchar(y) type x <= y") {
+    withTable("t") {
+      sql("CREATE TABLE t(i STRING, c VARCHAR(4)) USING delta")
+      sql("ALTER TABLE t CHANGE COLUMN c TYPE VARCHAR(4)")
+      checkColType(spark.table("t").schema(1), VarcharType(4))
+      sql("ALTER TABLE t CHANGE COLUMN c TYPE VARCHAR(5)")
+      checkColType(spark.table("t").schema(1), VarcharType(5))
+    }
+  }
 }
 
 trait DeltaAlterTableByNameTests extends DeltaAlterTableTests {