Support change column dataType

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

@@ -309,7 +309,7 @@ case class AlterTableChangeColumnCommand(
     columnName: String,
     newColumn: StructField) extends RunnableCommand {
 
-  // TODO: support change column name/dataType/metadata/position.
+  // TODO: support change column name/metadata/position.
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     val table = catalog.getTableMetadata(tableName)
@@ -326,10 +326,18 @@ case class AlterTableChangeColumnCommand(
           s"'${newColumn.name}' with type '${newColumn.dataType}'")
     }
 
-    val newDataSchema = table.dataSchema.fields.map { field =>
+    val changeSchema = originColumn.dataType != newColumn.dataType
+    val newDataSchema = table.schema.fields.map { field =>
       if (field.name == originColumn.name) {
-        // Create a new column from the origin column with the new comment.
-        addComment(field, newColumn.getComment)
+        var newField = field
+        if (newColumn.getComment.isDefined) {
+          // Create a new column from the origin column with the new comment.
+          newField = addComment(field, newColumn.getComment)
+        }
+        if (changeSchema) {
+          newField = newField.copy(dataType = newColumn.dataType)
+        }
+        newField
       } else {
         field
       }
@@ -359,7 +367,7 @@ case class AlterTableChangeColumnCommand(
   // name(by resolver) and dataType.
   private def columnEqual(
       field: StructField, other: StructField, resolver: Resolver): Boolean = {
-    resolver(field.name, other.name) && field.dataType == other.dataType
+    resolver(field.name, other.name)
   }
 }
 

sql/core/src/test/resources/sql-tests/inputs/change-column.sql

@@ -6,7 +6,7 @@ DESC test_change;
 ALTER TABLE test_change CHANGE a a1 INT;
 DESC test_change;
 
--- Change column dataType (not supported yet)
+-- Change column dataType
 ALTER TABLE test_change CHANGE a a STRING;
 DESC test_change;
 

sql/core/src/test/resources/sql-tests/results/change-column.sql.out

@@ -44,16 +44,15 @@ ALTER TABLE test_change CHANGE a a STRING
 -- !query 4 schema
 struct<>
 -- !query 4 output
-org.apache.spark.sql.AnalysisException
-ALTER TABLE CHANGE COLUMN is not supported for changing column 'a' with type 'IntegerType' to 'a' with type 'StringType';
+
 
 
 -- !query 5
 DESC test_change
 -- !query 5 schema
 struct<col_name:string,data_type:string,comment:string>
 -- !query 5 output
-a                   	int                 	                    
+a                   	string              	                    
 b                   	string              	                    
 c                   	int
 
@@ -91,7 +90,7 @@ DESC test_change
 -- !query 8 schema
 struct<col_name:string,data_type:string,comment:string>
 -- !query 8 output
-a                   	int                 	                    
+a                   	string              	                    
 b                   	string              	                    
 c                   	int
 

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -1697,6 +1697,11 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     sql("ALTER TABLE dbx.tab1 CHANGE COLUMN col1 col1 INT COMMENT 'this is col1'")
     assert(getMetadata("col1").getString("key") == "value")
     assert(getMetadata("col1").getString("comment") == "this is col1")
+
+    // Ensure that change column type take effect
+    sql("ALTER TABLE dbx.tab1 CHANGE COLUMN col1 col1 STRING")
+    val column = catalog.getTableMetadata(tableIdent).schema.fields.find(_.name == "col1")
+    assert(column.get.dataType == StringType)
   }
 
   test("drop build-in function") {