[CARMEL-5955] Drop NOT NULL constraint when alter column type (#930)

* [CARMEL-5955] Drop NOT NULL constraint when alter column type

* refresh

Author: fenzhu

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

@@ -542,15 +542,25 @@ case class AlterTableChangeColumnCommand(
             s"'${newColumn.name}' with type '${newColumn.dataType}'")
       }
 
+      var changeNullable = false
       val newDataSchema = table.dataSchema.fields.map { field =>
         if (field.name == originColumn.name) {
           // Create a new column from the origin column with the new comment.
-          addComment(field, newColumn.getComment)
+          val fieldWithComment = addComment(field, newColumn.getComment)
+          if (!fieldWithComment.nullable && newColumn.nullable) {
+            changeNullable = true
+            fieldWithComment.copy(nullable = true)
+          } else {
+            fieldWithComment
+          }
         } else {
           field
         }
       }
       catalog.alterTableDataSchema(tableName, StructType(newDataSchema))
+      if (changeNullable && sparkSession.sessionState.conf.enforceSchemaNotNull) {
+        catalog.refreshTable(tableName)
+      }
     }
 
     Seq.empty[Row]

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -743,6 +743,27 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
     }
   }
 
+  test("CARMEL-5955: Drop NOT NULL constraint when alter column type") {
+    withSQLConf(SQLConf.ENFORCE_SCHEMA_NOT_NULL.key -> "true") {
+      withTable("test") {
+        sql("create table test(id int not null, name string) using parquet")
+        val schema = spark.sessionState.catalog.getTableMetadata(TableIdentifier("test")).schema
+        assert(schema.fields(0).name.equals("id") && !schema.fields(0).nullable
+          && schema.fields(1).name.equals("name") && schema.fields(1).nullable)
+
+        val e1 = intercept[SparkException] {
+          sql("insert into test values (null, 'a')")
+        }
+        assert(Utils.findFirstCause(e1).getMessage.contains(
+          "NOT NULL constraint violated for column: id"))
+
+        sql("alter table test change column id id int")
+        sql("insert into test values (null, 'a')")
+        checkAnswer(sql("select * from test"), Seq(Row(null, "a")))
+      }
+    }
+  }
+
   test("inner join, no matches") {
     checkAnswer(
       sql(

sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala

@@ -1344,6 +1344,25 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter
     }
   }
 
+  test("CARMEL-5955: Drop NOT NULL constraint when alter column type") {
+    withSQLConf(SQLConf.ENFORCE_SCHEMA_NOT_NULL.key -> "true") {
+      withTable("test") {
+        sql("create table test(id int not null, name string)")
+        val schema = spark.sessionState.catalog.getTableMetadata(TableIdentifier("test")).schema
+        assert(schema.fields(0).name.equals("id") && !schema.fields(0).nullable
+          && schema.fields(1).name.equals("name") && schema.fields(1).nullable)
+
+        val e1 = intercept[SparkException] {
+          sql("insert into test values (null, 'a')")
+        }
+        assert(Utils.findFirstCause(e1).getMessage.contains(
+          "NOT NULL constraint violated for column: id"))
+        sql("alter table test change column id id int")
+        sql("insert into test values (null, 'a')")
+        checkAnswer(sql("select * from test"), Seq(Row(null, "a")))
+      }
+    }
+  }
 
   test("Data inserting into HiveTable violates table constraints") {
     withTable("constraintTable") {