[Spark] Fix a bug that prevents altering array/map/struct<varchar> column (delta-io#4499)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

### Context

Previously, there was a bug where char/varchar to string conversion
didn't work correctly due to how they are internally represented.

For example: 
- An `array<varchar<10>>` is represented as `(dataType = array<string>,
char_varchar_metadata = "array<varchar<10>>")`.
- If we convert `array<varchar<10>>` into `array<string>` by setting the
`dataType` part with`array<string>`, nothing will change.

This has been fixed in delta-io#3346. The
idea is to first convert the representation into one without the
metadata part. In the example above, `(dataType = array<string>,
char_varchar_metadata = "array<varchar<10>>")` will first be converted
into `(dataType = array<varchar<10>>, char_varchar_metadata = "")`
before setting the `dataType` part to `string`.

### Problem

In the previous fix, the non-metadata representation is left open and is
converted back to the metadata-based representation where necessary.
This is causing an issue where we incorrectly do this converting-back
step that breaks ALTER COLUMN on any columns with the type
`container<varchar>` where `container` can be `array`/`map`/`struct`.

Specifically, the `verifyColumnChange` is taking in `change` which is a
non-metadata-based representation, and a `oldColumnForVerification`,
which is a metadata-based representation. This method has a check to
prevent any change to complex data types (`array`/`map`/`struct`). Since
it's a simple equality check, the check fails when comparing the
non-metadata and metadata-based representation of the type.

### New Approach

To avoid having to reason about what representation to use where, this
PR takes a more targeted approach. It always defaults to the
metadata-based representation of char/varchar and only do the
representation conversion when setting the data type.

## How was this patch tested?

New and existing unit tests

## Does this PR introduce _any_ user-facing changes?

No

Author: ctring

spark/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala

@@ -747,14 +747,7 @@ class DeltaCatalog extends DelegatingCatalogExtension
 
       case (t, columnChanges) if classOf[ColumnChange].isAssignableFrom(t) =>
         // TODO: Theoretically we should be able to fetch the snapshot from a txn.
-        val snapshotSchema = table.initialSnapshot.schema
-        val schema = if (!spark.conf.get(DeltaSQLConf.DELTA_BYPASS_CHARVARCHAR_TO_STRING_FIX)) {
-          // Convert (StringType, metadata = 'VARCHAR(n)') into (VARCHAR(n), metadata = '')
-          // so that CHAR/VARCHAR to String conversion can be handled correctly.
-          SchemaUtils.getRawSchemaWithoutCharVarcharMetadata(snapshotSchema)
-        } else {
-          snapshotSchema
-        }
+        val schema = table.initialSnapshot.schema
         def getColumn(fieldNames: Seq[String])
             : DeltaChangeColumnSpec = {
           columnUpdates.getOrElseUpdate(fieldNames, {
@@ -800,8 +793,9 @@ class DeltaCatalog extends DelegatingCatalogExtension
           case dataType: UpdateColumnType =>
             val field = dataType.fieldNames()
             val spec = getColumn(field)
-            columnUpdates(field) = spec.copy(
-              newColumn = spec.newColumn.copy(dataType = dataType.newDataType()))
+            val newField = SchemaUtils.setFieldDataTypeCharVarcharSafe(
+              spec.newColumn, dataType.newDataType())
+            columnUpdates(field) = spec.copy(newColumn = newField)
 
           case position: UpdateColumnPosition =>
             val field = position.fieldNames()

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

@@ -765,11 +765,7 @@ case class AlterTableChangeColumnDeltaCommand(
       val metadata = txn.metadata
       val bypassCharVarcharToStringFix =
         sparkSession.conf.get(DeltaSQLConf.DELTA_BYPASS_CHARVARCHAR_TO_STRING_FIX)
-      val oldSchema = if (bypassCharVarcharToStringFix) {
-          metadata.schema
-        } else {
-          SchemaUtils.getRawSchemaWithoutCharVarcharMetadata(metadata.schema)
-        }
+      val oldSchema = metadata.schema
       val resolver = sparkSession.sessionState.conf.resolver
 
       columnChanges.foreach(change => {
@@ -792,16 +788,7 @@ case class AlterTableChangeColumnDeltaCommand(
         transformSchema(prevSchema, Some(columnName)) {
           case (`columnPath`, struct @ StructType(fields), _) =>
             val oldColumn = struct(columnName)
-
-            // Analyzer already validates the char/varchar type change of ALTER COLUMN in
-            // `CheckAnalysis.checkAlterTableCommand`. We should normalize char/varchar type
-            // to string type first, then apply Delta-specific checks.
-            val oldColumnForVerification = if (bypassCharVarcharToStringFix) {
-              oldColumn
-            } else {
-              CharVarcharUtils.replaceCharVarcharWithStringInSchema(StructType(Seq(oldColumn))).head
-            }
-            verifyColumnChange(change, sparkSession, oldColumnForVerification, resolver, txn)
+            verifyColumnChange(change, sparkSession, oldColumn, resolver, txn)
 
             val newField = {
               if (change.syncIdentity) {
@@ -828,13 +815,10 @@ case class AlterTableChangeColumnDeltaCommand(
                   case Some(newDefaultValue) => result.withCurrentDefaultValue(newDefaultValue)
                   case None => result.clearCurrentDefaultValue()
                 }
-
-                result
-                  .copy(
-                    name = newColumn.name,
-                    dataType =
-                      SchemaUtils.changeDataType(oldColumn.dataType, newColumn.dataType, resolver),
-                    nullable = newColumn.nullable)
+                result = SchemaUtils.changeFieldDataTypeCharVarcharSafe(result, newColumn, resolver)
+                result.copy(
+                  name = newColumn.name,
+                  nullable = newColumn.nullable)
               }
             }
 
@@ -854,19 +838,23 @@ case class AlterTableChangeColumnDeltaCommand(
           case (`columnPath`, m: MapType, _) if columnName == "key" =>
             val originalField = StructField(columnName, m.keyType, nullable = false)
             verifyMapArrayChange(change, sparkSession, originalField, resolver, txn)
-            m.copy(keyType = SchemaUtils.changeDataType(m.keyType, newColumn.dataType, resolver))
+            val fieldWithNewDataType = SchemaUtils.changeFieldDataTypeCharVarcharSafe(
+              originalField, newColumn, resolver)
+            m.copy(keyType = fieldWithNewDataType.dataType)
 
           case (`columnPath`, m: MapType, _) if columnName == "value" =>
             val originalField = StructField(columnName, m.valueType, nullable = m.valueContainsNull)
             verifyMapArrayChange(change, sparkSession, originalField, resolver, txn)
-            m.copy(
-              valueType = SchemaUtils.changeDataType(m.valueType, newColumn.dataType, resolver))
+            val fieldWithNewDataType = SchemaUtils.changeFieldDataTypeCharVarcharSafe(
+              originalField, newColumn, resolver)
+            m.copy(valueType = fieldWithNewDataType.dataType)
 
           case (`columnPath`, a: ArrayType, _) if columnName == "element" =>
             val originalField = StructField(columnName, a.elementType, nullable = a.containsNull)
             verifyMapArrayChange(change, sparkSession, originalField, resolver, txn)
-            a.copy(elementType =
-              SchemaUtils.changeDataType(a.elementType, newColumn.dataType, resolver))
+            val fieldWithNewDataType = SchemaUtils.changeFieldDataTypeCharVarcharSafe(
+              originalField, newColumn, resolver)
+            a.copy(elementType = fieldWithNewDataType.dataType)
 
           case (_, other @ (_: StructType | _: ArrayType | _: MapType), _) => other
         }

spark/src/main/scala/org/apache/spark/sql/delta/schema/SchemaUtils.scala

@@ -163,23 +163,68 @@ object SchemaUtils extends DeltaLogging {
   }
 
   /**
-   * Converts StringType to CHAR/VARCHAR if that is the true type as per the metadata
-   * and also strips this metadata from fields.
+   * A char(x)/varchar(x) related types are internally stored as string type with the constraint
+   * information stored in the metadata. For example:
+   *  + char(10) -> (string, char_varchar_metadata = "char(10)")
+   *  + array[varchar(10)]
+   *    -> (array[string], char_varchar_metadata = "array[varchar(10)]")
+   * This method converts the string + metadata representation to the actual type.
+   *  + (string, metadata = "char(10)") -> (char(10), char_varchar_metadata = "")
+   *  + (array[string], metadata = "array[varchar(10)]")
+   *    -> (array[varchar(10)], char_varchar_metadata = "")
    */
-  def getRawSchemaWithoutCharVarcharMetadata(schema: StructType): StructType = {
-    val fields = schema.map { field =>
-      val rawField = CharVarcharUtils.getRawType(field.metadata)
-        .map(dt => field.copy(dataType = dt))
-        .getOrElse(field)
-      val throwAwayAttrRef = AttributeReference(
-        rawField.name,
-        rawField.dataType,
-        nullable = rawField.nullable,
-        rawField.metadata)()
-      val cleanedMetadata = CharVarcharUtils.cleanAttrMetadata(throwAwayAttrRef).metadata
-      rawField.copy(metadata = cleanedMetadata)
+  private def getRawFieldWithoutCharVarcharMetadata(field: StructField): StructField = {
+    val rawField = CharVarcharUtils.getRawType(field.metadata)
+      .map(dt => field.copy(dataType = dt))
+      .getOrElse(field)
+    val throwAwayAttrRef = AttributeReference(
+      rawField.name,
+      rawField.dataType,
+      nullable = rawField.nullable,
+      rawField.metadata)()
+    val cleanedMetadata = CharVarcharUtils.cleanAttrMetadata(throwAwayAttrRef).metadata
+    rawField.copy(metadata = cleanedMetadata)
+  }
+
+  /**
+   * Sets a data type to a field in a char/varchar-safe manner. A char(x)/varchar(x) related types
+   * consists of two parts: a string-based type and the constraint information stored in the
+   * metadata. Simply setting the data type will lead to unexpected results.
+   *
+   * For example, an array[varchar(10)] type is internally represented as
+   * (array[string], char_varchar_metadata = "array[varchar(10)]"). If we convert it into a string
+   * simply by setting the data type part, the metadata part will still be there, and the type will
+   * still stay as varchar(10).
+   *
+   * This method first converts the field into its raw type without the metadata part, then sets the
+   * data type to the new data type, and finally converts it back to the original representation.
+   *
+   * In the above example, this methods will convert the array[varchar(10)] representation into
+   * (array[varchar(10)], char_varchar_metadata = ""), set the data type to string
+   * (string, char_varchar_metadata = ""), and finally convert it back to
+   * (string, char_varchar_metadata = ""), which happens to be the same.
+   */
+  def setFieldDataTypeCharVarcharSafe(field: StructField, newDataType: DataType): StructField = {
+    val byPassCharVarcharToStringFix =
+      SparkSession.active.conf.get(DeltaSQLConf.DELTA_BYPASS_CHARVARCHAR_TO_STRING_FIX)
+    // Convert the field into its raw type without the metadata part
+    val rawField = if (byPassCharVarcharToStringFix) {
+      field
+    } else {
+      getRawFieldWithoutCharVarcharMetadata(field)
+    }
+
+    // Set the new data type
+    val rawFieldWithNewDataType = rawField.copy(dataType = newDataType)
+
+    // Convert it back to the original representation
+    if (byPassCharVarcharToStringFix) {
+      rawFieldWithNewDataType
+    } else {
+      val throwAwayStructType = StructType(Seq(rawFieldWithNewDataType))
+      CharVarcharUtils.replaceCharVarcharWithStringInSchema(throwAwayStructType)
+        .head
     }
-    StructType(fields)
   }
 
   /**
@@ -1066,6 +1111,43 @@ def normalizeColumnNamesInDataType(
     }
   }
 
+  /**
+   * Copy the nested data type between two data types in a char/varchar safe manner.
+   * See documentation of [[getRawFieldWithoutCharVarcharMetadata]] and
+   * [[setFieldDataTypeCharVarcharSafe]] for more context.
+   *
+   * This method uses [[getRawFieldWithoutCharVarcharMetadata]] on both the source and
+   * target fields to ensure that the metadata information is included in the data type
+   * before changing the data type. For example, to convert from a varchar(1) to varchar(10),
+   * we first change their representation:
+   *
+   * Source: (string, char_varchar_metadata = "varchar(1)")
+   *  -> (varchar(1), char_varchar_metadata = "")
+   * Target: (string, char_varchar_metadata = "varchar(10)")
+   *  -> (varchar(10), char_varchar_metadata = "")
+   *
+   * Then, we change the data type of the target:
+   * (varchar(1), char_varchar_metadata = "") -> (varchar(10), char_varchar_metadata = "")
+   *
+   * Finally, we set the metadata back to the target field:
+   * (varchar(10), char_varchar_metadata = "") -> (string, char_varchar_metadata = "varchar(10)")
+   */
+  def changeFieldDataTypeCharVarcharSafe(
+      fromField: StructField,
+      toField: StructField,
+      resolver: Resolver): StructField = {
+    val (safeFromField, safeToField) =
+      if (SparkSession.active.conf.get(DeltaSQLConf.DELTA_BYPASS_CHARVARCHAR_TO_STRING_FIX)) {
+        (fromField, toField)
+      } else {
+        (getRawFieldWithoutCharVarcharMetadata(fromField),
+         getRawFieldWithoutCharVarcharMetadata(toField))
+      }
+    val newDataType = SchemaUtils.changeDataType(
+      safeFromField.dataType, safeToField.dataType, resolver)
+    setFieldDataTypeCharVarcharSafe(fromField, newDataType)
+  }
+
   /**
    * Copy the nested data type between two data types.
    */

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

@@ -1447,6 +1447,59 @@ trait DeltaAlterTableTests extends DeltaAlterTableTestBase {
     }
   }
 
+  ddlTest("CHANGE COLUMN - set comment on a array/map/struct<varchar> column") {
+    val schema = """
+      |arr_v array<varchar(1)>,
+      |map_vv map<varchar(1), varchar(1)>,
+      |map_sv map<string, varchar(1)>,
+      |map_vs map<varchar(1), string>,
+      |struct_v struct<v: varchar(1)>""".stripMargin
+    def testCommentOnVarcharInContainer(
+      colName: String,
+      expectedType: String,
+      goodInsertValue: String,
+      badInsertValue: String
+    ): Unit = {
+      withDeltaTable(schema = schema) { tableName =>
+        sql(s"ALTER TABLE $tableName CHANGE COLUMN $colName COMMENT 'test comment'")
+        val expectedResult = Row(colName, expectedType, "test comment") :: Nil
+        checkAnswer(
+          sql(s"DESCRIBE $tableName").filter(s"col_name = '$colName'"),
+          expectedResult)
+        sql(s"INSERT into $tableName($colName) values ($goodInsertValue)")
+        val e = intercept[DeltaInvariantViolationException] {
+          sql(s"INSERT into $tableName($colName) values ($badInsertValue)")
+        }
+        assert(e.getMessage.contains("exceeds char/varchar type length limitation"))
+      }
+    }
+    testCommentOnVarcharInContainer(
+      colName = "arr_v",
+      expectedType = "array<string>",
+      goodInsertValue = "array('1')",
+      badInsertValue = "array('12')")
+    testCommentOnVarcharInContainer(
+      colName = "map_vv",
+      expectedType = "map<string,string>",
+      goodInsertValue = "map('1', '1')",
+      badInsertValue = "map('12', '12')")
+    testCommentOnVarcharInContainer(
+      colName = "map_sv",
+      expectedType = "map<string,string>",
+      goodInsertValue = "map('123', '1')",
+      badInsertValue = "map('123', '12')")
+    testCommentOnVarcharInContainer(
+      colName = "map_vs",
+      expectedType = "map<string,string>",
+      goodInsertValue = "map('1', '123')",
+      badInsertValue = "map('12', '123')")
+    testCommentOnVarcharInContainer(
+      colName = "struct_v",
+      expectedType = "struct<v:string>",
+      goodInsertValue = "named_struct('v', '1')",
+      badInsertValue = "named_struct('v', '12')")
+  }
+
   ddlTest("CHANGE COLUMN - set a default value for a varchar column") {
     withDeltaTable(schema = "v varchar(1)") { tableName =>
       sql(s"ALTER TABLE $tableName " +