[SPARK-58794][SQL] Add standardSemantics foundation for CHAR/VARCHAR

sql/api/src/main/scala/org/apache/spark/sql/Encoders.scala

@@ -86,14 +86,14 @@ object Encoders {
    *
    * @since 4.0.0
    */
-  def CHAR(length: Int): Encoder[java.lang.String] = CharEncoder(length)
+  def CHAR(length: Int): Encoder[java.lang.String] = CharEncoder(CharType(length))
 
   /**
    * An encoder for nullable varchar type.
    *
    * @since 4.0.0
    */
-  def VARCHAR(length: Int): Encoder[java.lang.String] = VarcharEncoder(length)
+  def VARCHAR(length: Int): Encoder[java.lang.String] = VarcharEncoder(VarcharType(length))
 
   /**
    * An encoder for nullable string type.

sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/AgnosticEncoder.scala

@@ -237,8 +237,10 @@ object AgnosticEncoders {
   // Nullable leaf encoders
   case object NullEncoder extends LeafEncoder[java.lang.Void](NullType)
   case object StringEncoder extends LeafEncoder[String](StringType)
-  case class CharEncoder(length: Int) extends LeafEncoder[String](CharType(length))
-  case class VarcharEncoder(length: Int) extends LeafEncoder[String](VarcharType(length))
+  // Carry the full constrained type (length + collation), matching GeographyEncoder /
+  // GeometryEncoder. Reconstructing from length alone would drop a declared collation.
+  case class CharEncoder(dt: CharType) extends LeafEncoder[String](dt)
+  case class VarcharEncoder(dt: VarcharType) extends LeafEncoder[String](dt)
   case object BinaryEncoder extends LeafEncoder[Array[Byte]](BinaryType)
   case object ScalaBigIntEncoder extends LeafEncoder[BigInt](DecimalType.BigIntDecimal)
   case object JavaBigIntEncoder extends LeafEncoder[JBigInt](DecimalType.BigIntDecimal)

sql/api/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala

@@ -91,10 +91,10 @@ object RowEncoder extends DataTypeErrorsBase {
       case DoubleType => BoxedDoubleEncoder
       case dt: DecimalType => JavaDecimalEncoder(dt, lenientSerialization = true)
       case BinaryType => BinaryEncoder
-      case c: CharType if SqlApiConf.get.preserveCharVarcharTypeInfo =>
-        CharEncoder(c.length)
-      case v: VarcharType if SqlApiConf.get.preserveCharVarcharTypeInfo =>
-        VarcharEncoder(v.length)
+      case c: CharType if SqlApiConf.get.charVarcharFirstClassTypes =>
+        CharEncoder(c)
+      case v: VarcharType if SqlApiConf.get.charVarcharFirstClassTypes =>
+        VarcharEncoder(v)
       case s: StringType if StringHelper.isPlainString(s) => StringEncoder
       case TimestampType if SqlApiConf.get.datetimeJava8ApiEnabled => InstantEncoder(lenient)
       case TimestampType => TimestampEncoder(lenient)

sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkCharVarcharUtils.scala

@@ -31,11 +31,13 @@ trait SparkCharVarcharUtils {
   }
 
   /**
-   * Validate the given [[DataType]] to fail if it is char or varchar types or contains nested
-   * ones
+   * Fail if the type contains CHAR/VARCHAR unless legacy-as-string or first-class CHAR/VARCHAR is
+   * enabled (standard semantics or preserveCharVarcharTypeInfo).
    */
   def failIfHasCharVarchar(dt: DataType): DataType = {
-    if (!SqlApiConf.get.charVarcharAsString && hasCharVarchar(dt)) {
+    if (SqlApiConf.get.charVarcharFirstClassTypes) {
+      dt
+    } else if (!SqlApiConf.get.charVarcharAsString && hasCharVarchar(dt)) {
       throw DataTypeErrors.charOrVarcharTypeAsStringUnsupportedError()
     } else {
       replaceCharVarcharWithString(dt)
@@ -54,8 +56,8 @@ trait SparkCharVarcharUtils {
       StructType(fields.map { field =>
         field.copy(dataType = replaceCharVarcharWithString(field.dataType))
       })
-    case c: CharType if !SqlApiConf.get.preserveCharVarcharTypeInfo => c.toStringType
-    case v: VarcharType if !SqlApiConf.get.preserveCharVarcharTypeInfo => v.toStringType
+    case c: CharType if !SqlApiConf.get.charVarcharFirstClassTypes => c.toStringType
+    case v: VarcharType if !SqlApiConf.get.charVarcharFirstClassTypes => v.toStringType
     case _ => dt
   }
 }

sql/api/src/main/scala/org/apache/spark/sql/internal/SqlApiConf.scala

@@ -42,6 +42,14 @@ private[sql] trait SqlApiConf {
   def allowNegativeScaleOfDecimalEnabled: Boolean
   def charVarcharAsString: Boolean
   def preserveCharVarcharTypeInfo: Boolean
+  def charVarcharStandardSemantics: Boolean
+
+  /**
+   * True when CHAR/VARCHAR may appear as first-class types in schemas and plans (either the
+   * legacy preserve path or SQL standard semantics).
+   */
+  def charVarcharFirstClassTypes: Boolean =
+    preserveCharVarcharTypeInfo || charVarcharStandardSemantics
   def datetimeJava8ApiEnabled: Boolean
   def sessionLocalTimeZone: String
   def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value
@@ -104,6 +112,7 @@ private[sql] object DefaultSqlApiConf extends SqlApiConf {
   override def allowNegativeScaleOfDecimalEnabled: Boolean = false
   override def charVarcharAsString: Boolean = false
   override def preserveCharVarcharTypeInfo: Boolean = false
+  override def charVarcharStandardSemantics: Boolean = false
   override def datetimeJava8ApiEnabled: Boolean = false
   override def sessionLocalTimeZone: String = TimeZone.getDefault.getID
   override def legacyTimeParserPolicy: LegacyBehaviorPolicy.Value = LegacyBehaviorPolicy.CORRECTED

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

@@ -159,25 +159,76 @@ case object StringHelper extends PartialOrdering[StringConstraint] {
 
   def isPlainString(s: StringType): Boolean = s.constraint == NoConstraint
 
+  /**
+   * 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.
+   */
+  def plainStringType(dt: DataType): DataType = dt match {
+    case c: CharType => c.toStringType
+    case v: VarcharType => v.toStringType
+    case other => other
+  }
+
+  def plainStringType(s: StringType): StringType = s match {
+    case c: CharType => c.toStringType
+    case v: VarcharType => v.toStringType
+    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)
 
+  /**
+   * Least common string type: CHAR -> VARCHAR -> STRING, with length max(n, m) when the result
+   * remains CHAR or VARCHAR.
+   *
+   * When first-class CHAR/VARCHAR are off, always widens to unbounded STRING (legacy
+   * annotated-STRING path where Char/Varchar do not appear in plans).
+   */
   def tightestCommonString(s1: StringType, s2: StringType): Option[StringType] = {
     if (s1.collationId != s2.collationId) {
       return None
     }
-    if (!SqlApiConf.get.preserveCharVarcharTypeInfo) {
+    if (!SqlApiConf.get.charVarcharFirstClassTypes) {
       return Some(StringType(s1.collationId))
     }
+    // Carry the declared collation onto CHAR/VARCHAR results. This propagates the Option rather
+    // than the id: None means "not explicitly declared" and renders as char(n) instead of
+    // char(n) collate UTF8_BINARY, so an all-default LCT keeps printing (and comparing) as before.
+    // The two collation ids are already known to be equal here.
+    val collation = declaredCollation(s1).orElse(declaredCollation(s2))
     Some((s1.constraint, s2.constraint) match {
-      case (FixedLength(l1), FixedLength(l2)) => CharType(l1.max(l2))
-      case (MaxLength(l1), FixedLength(l2)) => VarcharType(l1.max(l2))
-      case (FixedLength(l1), MaxLength(l2)) => VarcharType(l1.max(l2))
-      case (MaxLength(l1), MaxLength(l2)) => VarcharType(l1.max(l2))
+      case (FixedLength(l1), FixedLength(l2)) => new CharType(l1.max(l2), collation)
+      case (MaxLength(l1), FixedLength(l2)) => new VarcharType(l1.max(l2), collation)
+      case (FixedLength(l1), MaxLength(l2)) => new VarcharType(l1.max(l2), collation)
+      case (MaxLength(l1), MaxLength(l2)) => new VarcharType(l1.max(l2), collation)
       case _ => StringType(s1.collationId)
     })
   }
 
+  /** The explicitly declared collation of a CHAR/VARCHAR type, if any. */
+  private def declaredCollation(s: StringType): Option[Int] = s match {
+    case c: CharType => c.collation
+    case v: VarcharType => v.collation
+    case _ => None
+  }
+
   def removeCollation(s: StringType): StringType = s match {
     case c: CharType => CharType(c.length)
     case v: VarcharType => VarcharType(v.length)

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/util/CharVarcharCodegenUtils.java

@@ -64,4 +64,27 @@ public static UTF8String readSidePadding(UTF8String inputStr, int limit) {
       return inputStr;
     }
   }
+
+  /**
+   * Read-side CHAR check under standard semantics: pad to limit, or trim trailing
+   * spaces then error if still longer than limit.
+   *
+   * Standard semantics require a read to observe the same value a write would have
+   * produced, so this is deliberately the write-side check rather than
+   * {@link #readSidePadding}, which tolerates over-long values. Keep the two sides
+   * identical: a fix to one is a fix to both.
+   */
+  public static UTF8String charTypeReadSideCheck(UTF8String inputStr, int limit) {
+    return charTypeWriteSideCheck(inputStr, limit);
+  }
+
+  /**
+   * Read-side VARCHAR check under standard semantics: allow up to limit characters,
+   * or trim trailing spaces then error if still longer than limit.
+   *
+   * Identical to the write-side check by design; see {@link #varcharTypeWriteSideCheck}.
+   */
+  public static UTF8String varcharTypeReadSideCheck(UTF8String inputStr, int limit) {
+    return varcharTypeWriteSideCheck(inputStr, limit);
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala

@@ -327,10 +327,10 @@ object DeserializerBuildHelper {
       createDeserializerForGeographyType(path, g.dt)
     case g: GeometryEncoder =>
       createDeserializerForGeometryType(path, g.dt)
-    case CharEncoder(length) =>
-      createDeserializerForChar(path, returnNullable = false, length)
-    case VarcharEncoder(length) =>
-      createDeserializerForVarchar(path, returnNullable = false, length)
+    case CharEncoder(dt) =>
+      createDeserializerForChar(path, returnNullable = false, dt.length)
+    case VarcharEncoder(dt) =>
+      createDeserializerForVarchar(path, returnNullable = false, dt.length)
     case StringEncoder =>
       createDeserializerForString(path, returnNullable = false)
     case _: ScalaDecimalEncoder =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala

@@ -358,8 +358,8 @@ object SerializerBuildHelper {
         messageParameters = scala.collection.immutable.Map.empty)
     case g: GeographyEncoder => createSerializerForGeographyType(input, g.dt)
     case g: GeometryEncoder => createSerializerForGeometryType(input, g.dt)
-    case CharEncoder(length) => createSerializerForChar(input, length)
-    case VarcharEncoder(length) => createSerializerForVarchar(input, length)
+    case CharEncoder(dt) => createSerializerForChar(input, dt.length)
+    case VarcharEncoder(dt) => createSerializerForVarchar(input, dt.length)
     case StringEncoder => createSerializerForString(input)
     case ScalaDecimalEncoder(dt) => createSerializerForBigDecimal(input, dt)
     case JavaDecimalEncoder(dt, false) => createSerializerForBigDecimal(input, dt)

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

@@ -168,6 +168,10 @@ 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
+    // accept the constrained type unchanged, since CharType and VarcharType extend StringType.
+    charVarcharToPlainString(inType, expectedType).foreach(dt => return Some(dt))
+
     (inType, expectedType) match {
       // If the expected type equals the input type, no need to cast.
       case _ if expectedType.acceptsType(inType) => Some(inType)

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

@@ -363,11 +363,13 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
     plan.foreachUp {
       case p if p.analyzed => // Skip already analyzed sub-plans
 
-      case leaf: LeafNode if !SQLConf.get.preserveCharVarcharTypeInfo &&
-        leaf.output.map(_.dataType).exists(CharVarcharUtils.hasCharVarchar) =>
+      case leaf: LeafNode
+          if !SQLConf.get.charVarcharFirstClassTypes &&
+            leaf.output.exists(attr => CharVarcharUtils.hasCharVarchar(attr.dataType)) =>
         throw SparkException.internalError(
           s"Logical plan should not have output of char/varchar type when " +
-            s"${SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key} is false: " + leaf)
+            s"${SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key} and " +
+            s"${SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key} are both false: " + leaf)
 
       case u: UnresolvedNamespace =>
         u.schemaNotFound(u.multipartIdentifier)

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

@@ -143,7 +143,7 @@ class ResolveCatalogs(val catalogManager: CatalogManager)
       allowTemp: Boolean,
       columns: Seq[ColumnDefinition]): ResolvedIdentifier = {
     val columnOutput = columns.map { col =>
-      val dataType = if (conf.preserveCharVarcharTypeInfo) {
+      val dataType = if (conf.charVarcharFirstClassTypes) {
         col.dataType
       } else {
         replaceCharVarcharWithString(col.dataType)

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

@@ -202,6 +202,10 @@ 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
+    // accept the constrained type unchanged, since CharType and VarcharType extend StringType.
+    charVarcharToPlainString(inType, expectedType).foreach(dt => return Some(dt))
+
     @Nullable val ret: DataType = (inType, expectedType) match {
       // If the expected type is already a parent of the input type, no need to cast.
       case _ if expectedType.acceptsType(inType) => inType

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

@@ -51,10 +51,54 @@ 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.types.DataType
+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
+}
 
 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/expressions/collectionOperations.scala

@@ -1430,7 +1430,11 @@ case class Reverse(child: Expression)
       BinaryType,
       ArrayType))
 
-  override def dataType: DataType = child.dataType
+  // 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)
 
   private def resultArrayElementNullable = dataType.asInstanceOf[ArrayType].containsNull
 
@@ -2424,7 +2428,10 @@ case class ArrayJoin(
     }
   }
 
-  override def dataType: DataType = array.dataType.asInstanceOf[ArrayType].elementType
+  // The joined result concatenates every element plus delimiters, so it must not inherit the
+  // element's CHAR/VARCHAR length constraint (R1).
+  override def dataType: DataType =
+    StringHelper.transformingStringResultType(array.dataType.asInstanceOf[ArrayType].elementType)
 
   override def prettyName: String = "array_join"
 
@@ -3111,7 +3118,7 @@ case class Concat(children: Seq[Expression]) extends ComplexTypeMergingExpressio
     if (children.isEmpty) {
       StringType
     } else {
-      super.dataType
+      StringHelper.transformingStringResultType(super.dataType)
     }
   }
 

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

@@ -640,9 +640,14 @@ case class StringToMap(text: Expression, pairDelim: Expression, keyValueDelim: E
   override def inputTypes: Seq[AbstractDataType] =
     Seq(StringTypeNonCSAICollation, StringTypeNonCSAICollation, StringTypeNonCSAICollation)
 
-  override def dataType: DataType = MapType(first.dataType, first.dataType)
+  // 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.
+  private lazy val entryType: DataType =
+    StringHelper.transformingStringResultType(first.dataType)
 
-  private lazy val mapBuilder = new ArrayBasedMapBuilder(first.dataType, first.dataType)
+  override def dataType: DataType = MapType(entryType, entryType)
+
+  private lazy val mapBuilder = new ArrayBasedMapBuilder(entryType, entryType)
 
   private final lazy val collationId: Int = text.dataType.asInstanceOf[StringType].collationId