[SPARK-58794][SQL] Promote CHAR/VARCHAR to STRING via ImplicitTypeCasts

sql/api/src/main/scala/org/apache/spark/sql/types/StringType.scala

@@ -163,8 +163,7 @@ case object StringHelper extends PartialOrdering[StringConstraint] {
    * Strip CHAR/VARCHAR length constraints, preserving collation.
    *
    * Used by transforming string expressions (upper, substr, concat, ...) so their result type is
-   * plain STRING even when inputs are CharType/VarcharType (SQL standard CHAR/VARCHAR R1), when
-   * standard semantics are on.
+   * plain STRING even when inputs are CharType/VarcharType, when standard semantics are on.
    */
   def plainStringType(dt: DataType): DataType = dt match {
     case c: CharType => c.toStringType
@@ -178,19 +177,6 @@ case object StringHelper extends PartialOrdering[StringConstraint] {
     case other => other
   }
 
-  /**
-   * Result type for transforming string expressions. Under
-   * spark.sql.charVarchar.standardSemantics.enabled, always plain STRING (R1). Under
-   * preserveCharVarcharTypeInfo alone, keep child type (legacy leaky path).
-   */
-  def transformingStringResultType(dt: DataType): DataType = {
-    if (SqlApiConf.get.charVarcharStandardSemantics) {
-      plainStringType(dt)
-    } else {
-      dt
-    }
-  }
-
   def isMoreConstrained(a: StringType, b: StringType): Boolean =
     gteq(a.constraint, b.constraint)
 

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

@@ -168,7 +168,7 @@ object AnsiTypeCoercion extends TypeCoercionBase {
   private def implicitCast(
       inType: DataType,
       expectedType: AbstractDataType): Option[DataType] = {
-    // R1 CHAR/VARCHAR promotion is checked first: the acceptsType case below would otherwise
+    // CHAR/VARCHAR promotion is checked first: the acceptsType case below would otherwise
     // accept the constrained type unchanged, since CharType and VarcharType extend StringType.
     charVarcharToPlainString(inType, expectedType).foreach(dt => return Some(dt))
 

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

@@ -202,7 +202,7 @@ object TypeCoercion extends TypeCoercionBase {
   private def implicitCast(inType: DataType, expectedType: AbstractDataType): Option[DataType] = {
     // Note that ret is nullable to avoid typing a lot of Some(...) in this local scope.
     // We wrap immediately an Option after this.
-    // R1 CHAR/VARCHAR promotion is checked first: the acceptsType case below would otherwise
+    // CHAR/VARCHAR promotion is checked first: the acceptsType case below would otherwise
     // accept the constrained type unchanged, since CharType and VarcharType extend StringType.
     charVarcharToPlainString(inType, expectedType).foreach(dt => return Some(dt))
 

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

@@ -51,54 +51,10 @@ import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
 import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure
 import org.apache.spark.sql.errors.DataTypeErrors.cannotMergeIncompatibleDataTypesError
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.types.AbstractStringType
-import org.apache.spark.sql.types.{
-  AbstractDataType,
-  DataType,
-  StringHelper,
-  StringType,
-  TypeCollection
-}
+import org.apache.spark.sql.types.DataType
 
 abstract class TypeCoercionBase extends TypeCoercionHelper {
 
-  /**
-   * R1 promotion for CHAR(n)/VARCHAR(n): where a plain string is expected, promote to STRING the
-   * same way SHORT promotes to INT, and return the promoted type.
-   *
-   * CharType and VarcharType extend StringType, so an expectation such as
-   * `StringTypeWithCollation` accepts them as-is and the implicit cast rules leave the length
-   * constraint in place. Expressions that then require all their string inputs to share a single
-   * type (`overlay`, `string_agg`, ...) cannot unify CHAR(n) with STRING, and RuntimeReplaceable
-   * ones (`right`) build literals from the constrained type that no longer match their other
-   * branches.
-   *
-   * The expectation must actually mention a string type. Promoting at an `AnyDataType` site would
-   * strip the length from pass-through expressions such as `max`, `lag`, and `element_at`, which
-   * are required to preserve CHAR/VARCHAR (R2/R3).
-   */
-  protected def charVarcharToPlainString(
-      inType: DataType,
-      expectedType: AbstractDataType): Option[DataType] = inType match {
-    case st: StringType
-        if SQLConf.get.charVarcharStandardSemantics && !StringHelper.isPlainString(st) =>
-      val plain = StringHelper.plainStringType(st)
-      if (expectsStringType(expectedType) && expectedType.acceptsType(plain)) {
-        Some(plain)
-      } else {
-        None
-      }
-    case _ => None
-  }
-
-  private def expectsStringType(expectedType: AbstractDataType): Boolean = expectedType match {
-    case _: StringType => true
-    case _: AbstractStringType => true
-    case TypeCollection(types) => types.exists(expectsStringType)
-    case _ => false
-  }
-
   /**
    * Type coercion rule that combines multiple type coercion rules and applies them in a single tree
    * traversal.

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

@@ -39,6 +39,7 @@ import org.apache.spark.sql.catalyst.expressions.{
   ImplicitCastInputTypes,
   In,
   InSubquery,
+  JsonTuple,
   Least,
   ListQuery,
   Literal,
@@ -60,7 +61,11 @@ import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Project}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.types.{AbstractArrayType, StringTypeWithCollation}
+import org.apache.spark.sql.internal.types.{
+  AbstractArrayType,
+  AbstractStringType,
+  StringTypeWithCollation
+}
 import org.apache.spark.sql.types.{
   AbstractDataType,
   AnyDataType,
@@ -80,6 +85,7 @@ import org.apache.spark.sql.types.{
   IntegralType,
   MapType,
   NullType,
+  StringHelper,
   StringType,
   StringTypeExpression,
   StructType,
@@ -89,7 +95,8 @@ import org.apache.spark.sql.types.{
   TimestampNTZType,
   TimestampType,
   TimestampTypeExpression,
-  TimeType
+  TimeType,
+  TypeCollection
 }
 
 abstract class TypeCoercionHelper {
@@ -133,6 +140,69 @@ abstract class TypeCoercionHelper {
    */
   def implicitCast(e: Expression, expectedType: AbstractDataType): Option[Expression]
 
+  /**
+   * Where a plain string is expected, promote CHAR(n)/VARCHAR(n) to unbounded STRING the same way
+   * SHORT promotes to INT, and return the promoted type.
+   *
+   * CharType and VarcharType extend StringType, so an expectation such as
+   * `StringTypeWithCollation` accepts them as-is and the implicit cast rules leave the length
+   * constraint in place. Expressions that then require all their string inputs to share a single
+   * type (`overlay`, `string_agg`, ...) cannot unify CHAR(n) with STRING, and RuntimeReplaceable
+   * ones (`right`) build literals from the constrained type that no longer match their other
+   * branches.
+   *
+   * The expectation must actually mention a string type (or an array of strings). Promoting at an
+   * `AnyDataType` site would strip the length from pass-through expressions such as `max`, `lag`,
+   * and `element_at`, which are required to preserve CHAR/VARCHAR.
+   */
+  protected def charVarcharToPlainString(
+      inType: DataType,
+      expectedType: AbstractDataType): Option[DataType] = {
+    if (!conf.charVarcharStandardSemantics) {
+      return None
+    }
+    inType match {
+      case st: StringType if !StringHelper.isPlainString(st) =>
+        val plain = StringHelper.plainStringType(st)
+        if (expectsStringType(expectedType) && expectedType.acceptsType(plain)) {
+          Some(plain)
+        } else {
+          None
+        }
+      case ArrayType(et, containsNull) =>
+        arrayElementExpectation(expectedType).flatMap { elemExpected =>
+          charVarcharToPlainString(et, elemExpected).map(ArrayType(_, containsNull))
+        }
+      case _ => None
+    }
+  }
+
+  private def expectsStringType(expectedType: AbstractDataType): Boolean = expectedType match {
+    case _: StringType => true
+    case _: AbstractStringType => true
+    case TypeCollection(types) => types.exists(expectsStringType)
+    case _ => false
+  }
+
+  private def arrayElementExpectation(expectedType: AbstractDataType): Option[AbstractDataType] =
+    expectedType match {
+      case AbstractArrayType(elem) => Some(elem)
+      case ArrayType(elem, _) => Some(elem)
+      case TypeCollection(types) => types.view.flatMap(arrayElementExpectation).headOption
+      case _ => None
+    }
+
+  /**
+   * Concat/Elt stringify non-binary inputs. CHAR/VARCHAR go through
+   * [[charVarcharToPlainString]] so a collated constrained type promotes to unbounded STRING
+   * with the same collation. Non-strings still target the default UTF8_BINARY StringType.
+   */
+  protected def implicitCastToString(e: Expression): Expression = {
+    charVarcharToPlainString(e.dataType, StringTypeWithCollation(supportsTrimCollation = true))
+      .map(dt => if (dt == e.dataType) e else Cast(e, dt))
+      .getOrElse(implicitCast(e, StringType).getOrElse(e))
+  }
+
   /**
    * Whether casting `from` as `to` is valid.
    */
@@ -512,9 +582,7 @@ abstract class TypeCoercionHelper {
       case c @ Concat(children)
           if conf.concatBinaryAsString ||
           !children.map(_.dataType).forall(_ == BinaryType) =>
-        val newChildren = c.children.map { e =>
-          implicitCast(e, StringType).getOrElse(e)
-        }
+        val newChildren = c.children.map(implicitCastToString)
         c.copy(children = newChildren)
       case other => other
     }
@@ -562,9 +630,7 @@ abstract class TypeCoercionHelper {
         val newInputs =
           if (conf.eltOutputAsString ||
             !children.tail.map(_.dataType).forall(_ == BinaryType)) {
-            children.tail.map { e =>
-              implicitCast(e, StringType).getOrElse(e)
-            }
+            children.tail.map(implicitCastToString)
           } else {
             children.tail
           }
@@ -649,16 +715,35 @@ abstract class TypeCoercionHelper {
         }
         e.withNewChildren(children)
 
+      // JsonTuple validates its own input types and rejects non-string children with
+      // NON_STRING_TYPE, so it only takes the CHAR/VARCHAR promotion here. Do not fold this into
+      // the ExpectsInputTypes arm below: that would also apply the NullType rewrite and turn
+      // json_tuple(json, null) from an analysis error into a typed STRING null.
+      case j: JsonTuple =>
+        val expected = StringTypeWithCollation(supportsTrimCollation = true)
+        val children = j.children.map { child =>
+          charVarcharToPlainString(child.dataType, expected)
+            .map(dt => if (dt == child.dataType) child else Cast(child, dt))
+            .getOrElse(child)
+        }
+        j.withNewChildren(children)
+
       case e: ExpectsInputTypes if e.inputTypes.nonEmpty =>
         // Convert NullType into some specific target type for ExpectsInputTypes that don't do
-        // general implicit casting.
+        // general implicit casting. Also promote CHAR/VARCHAR to STRING here: these
+        // expressions skip ImplicitCastInputTypes, so without this the length constraint would
+        // remain on the child.
         val children: Seq[Expression] = e.children.zip(e.inputTypes).map {
           case (in, expected) =>
-            if (in.dataType == NullType && !expected.acceptsType(NullType)) {
-              Literal.create(null, expected.defaultConcreteType)
-            } else {
-              in
-            }
+            charVarcharToPlainString(in.dataType, expected)
+              .map(dt => if (dt == in.dataType) in else Cast(in, dt))
+              .getOrElse {
+                if (in.dataType == NullType && !expected.acceptsType(NullType)) {
+                  Literal.create(null, expected.defaultConcreteType)
+                } else {
+                  in
+                }
+              }
         }
         e.withNewChildren(children)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala

@@ -1430,11 +1430,9 @@ case class Reverse(child: Expression)
       BinaryType,
       ArrayType))
 
-  // Reversing a string transforms its content, so a CHAR/VARCHAR input yields plain STRING (R1).
-  // Array and binary inputs are unaffected. ImplicitTypeCasts already promotes the string branch
-  // (its promotion looks inside a TypeCollection), so this covers the paths that do not go
-  // through implicit casting, such as an expression built directly.
-  override def dataType: DataType = StringHelper.transformingStringResultType(child.dataType)
+  // Reversing a string transforms its content, so ImplicitTypeCasts promotes CHAR/VARCHAR to
+  // STRING. Array and binary inputs are unaffected.
+  override def dataType: DataType = child.dataType
 
   private def resultArrayElementNullable = dataType.asInstanceOf[ArrayType].containsNull
 
@@ -2513,10 +2511,9 @@ case class ArrayJoin(
     }
   }
 
-  // The joined result concatenates every element plus delimiters, so it must not inherit the
-  // element's CHAR/VARCHAR length constraint (R1).
+  // After ImplicitTypeCasts, array elements that were CHAR/VARCHAR are STRING.
   override def dataType: DataType =
-    StringHelper.transformingStringResultType(array.dataType.asInstanceOf[ArrayType].elementType)
+    array.dataType.asInstanceOf[ArrayType].elementType
 
   override def prettyName: String = "array_join"
 
@@ -3203,7 +3200,7 @@ case class Concat(children: Seq[Expression]) extends ComplexTypeMergingExpressio
     if (children.isEmpty) {
       StringType
     } else {
-      StringHelper.transformingStringResultType(super.dataType)
+      super.dataType
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypeCreator.scala

@@ -641,9 +641,9 @@ case class StringToMap(text: Expression, pairDelim: Expression, keyValueDelim: E
     Seq(StringTypeNonCSAICollation, StringTypeNonCSAICollation, StringTypeNonCSAICollation)
 
   // The entries are split out of the input, so they do not carry its CHAR(n)/VARCHAR(n) length
-  // (R1). ExpectsInputTypes does not insert a cast, so the constraint has to be dropped here.
+  // constraint. ImplicitTypeCasts promotes CHAR/VARCHAR to STRING at this ExpectsInputTypes site.
   private lazy val entryType: DataType =
-    StringHelper.transformingStringResultType(first.dataType)
+    first.dataType
 
   override def dataType: DataType = MapType(entryType, entryType)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala

@@ -287,9 +287,10 @@ case class JsonTuple(children: Seq[Expression])
   }
 
   // The extracted fields are values from inside the JSON document, so they do not carry the
-  // CHAR(n)/VARCHAR(n) length of the document itself (R1).
+  // CHAR(n)/VARCHAR(n) length of the document itself. ImplicitTypeCoercion promotes CHAR/VARCHAR
+  // children to STRING without applying general implicit casts or rewriting untyped NULL.
   private lazy val fieldType: DataType =
-    StringHelper.transformingStringResultType(children.head.dataType)
+    children.head.dataType
 
   override def elementSchema: StructType = StructType(fieldExpressions.zipWithIndex.map {
     case (_, idx) => StructField(s"c$idx", fieldType, nullable = true)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/maskExpressions.scala

@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.codegen.Block._
 import org.apache.spark.sql.catalyst.plans.logical.{FunctionSignature, InputParameter}
 import org.apache.spark.sql.errors.QueryErrorsBase
 import org.apache.spark.sql.internal.types.StringTypeWithCollation
-import org.apache.spark.sql.types.{AbstractDataType, DataType, StringHelper, StringType}
+import org.apache.spark.sql.types.{AbstractDataType, DataType, StringType}
 import org.apache.spark.unsafe.types.UTF8String
 
 // scalastyle:off line.size.limit
@@ -291,7 +291,7 @@ case class Mask(
    * the dataType of an unresolved expression (i.e., when `resolved` == false).
    */
   override def dataType: DataType =
-    StringHelper.transformingStringResultType(input.dataType)
+    input.dataType
 
   /**
    * Returns a Seq of the children of this node. Children should not change. Immutability required

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala

@@ -1282,9 +1282,9 @@ case class Hex(child: Expression)
     Seq(TypeCollection(LongType, BinaryType, StringTypeWithCollation(supportsTrimCollation = true)))
 
   override def dataType: DataType = child.dataType match {
-    // Hex expands each input character to two hex digits, so a CHAR(n)/VARCHAR(n) input must not
-    // carry its length constraint into the result (R1).
-    case st: StringType => StringHelper.transformingStringResultType(st)
+    // After ImplicitTypeCasts, a CHAR/VARCHAR input is STRING. Keep collation from the
+    // promoted child rather than DefaultStringProducingExpression's UTF8_BINARY StringType.
+    case st: StringType => st
     case _ => super.dataType
   }