[SPARK-58807][SQL] Preserve CHAR/VARCHAR on language surfaces under standardSemantics

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

@@ -30,6 +30,16 @@ trait SparkCharVarcharUtils {
     dt.existsRecursively(f => f.isInstanceOf[CharType] || f.isInstanceOf[VarcharType])
   }
 
+  /**
+   * Logical type name for CHAR/VARCHAR, used when stamping `spark.sql.catalyst.type` on STRING
+   * storage (ORC/Avro). None for other types, including unbounded STRING.
+   */
+  def charVarcharTypeName(dt: DataType): Option[String] = dt match {
+    case c: CharType => Some(c.typeName)
+    case v: VarcharType => Some(v.typeName)
+    case _ => None
+  }
+
   /**
    * Fail if the type contains CHAR/VARCHAR unless legacy-as-string or first-class CHAR/VARCHAR is
    * enabled (standard semantics or preserveCharVarcharTypeInfo).

sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala

@@ -250,7 +250,7 @@ private[sql] class AvroDeserializer(
       case (DOUBLE, DoubleType) => (updater, ordinal, value) =>
         updater.setDouble(ordinal, value.asInstanceOf[Double])
 
-      case (STRING, StringType) => (updater, ordinal, value) =>
+      case (STRING, _: StringType) => (updater, ordinal, value) =>
         val str = value match {
           case s: String => UTF8String.fromString(s)
           case s: Utf8 =>
@@ -260,7 +260,7 @@ private[sql] class AvroDeserializer(
         }
         updater.set(ordinal, str)
 
-      case (ENUM, StringType) => (updater, ordinal, value) =>
+      case (ENUM, _: StringType) => (updater, ordinal, value) =>
         updater.set(ordinal, UTF8String.fromString(value.toString))
 
       case (FIXED, BinaryType) => (updater, ordinal, value) =>
@@ -345,8 +345,8 @@ private[sql] class AvroDeserializer(
 
           updater.set(ordinal, result)
 
-      case (MAP, MapType(keyType, valueType, valueContainsNull)) if keyType == StringType =>
-        val keyWriter = newWriter(SchemaBuilder.builder().stringType(), StringType,
+      case (MAP, MapType(keyType: StringType, valueType, valueContainsNull)) =>
+        val keyWriter = newWriter(SchemaBuilder.builder().stringType(), keyType,
           avroPath :+ "key", catalystPath :+ "key")
         val valueWriter = newWriter(avroType.getValueType, valueType,
           avroPath :+ "value", catalystPath :+ "value")

sql/core/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala

@@ -137,7 +137,7 @@ private[sql] class AvroSerializer(
           decimalConversions.toBytes(decimal.toJavaBigDecimal, avroType,
             LogicalTypes.decimal(d.precision, d.scale))
 
-      case (StringType, ENUM) =>
+      case (_: StringType, ENUM) =>
         val enumSymbols: Set[String] = avroType.getEnumSymbols.asScala.toSet
         (getter, ordinal) =>
           val data = getter.getUTF8String(ordinal).toString
@@ -148,7 +148,7 @@ private[sql] class AvroSerializer(
           }
           new EnumSymbol(avroType, data)
 
-      case (StringType, STRING) =>
+      case (_: StringType, STRING) =>
         (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes)
 
       case (BinaryType, FIXED) =>
@@ -259,7 +259,7 @@ private[sql] class AvroSerializer(
       case (LongType, UNION) if nonNullUnionTypes(avroType) == Set(INT, LONG) =>
         (getter, ordinal) => getter.getLong(ordinal)
 
-      case (MapType(kt, vt, valueContainsNull), MAP) if kt == StringType =>
+      case (MapType(_: StringType, vt, valueContainsNull), MAP) =>
         val valueConverter = newConverter(
           vt, resolveNullableType(avroType.getValueType, valueContainsNull),
           catalystPath :+ "value", avroPath :+ "value")

sql/core/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala

@@ -32,6 +32,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys.{FIELD_NAME, FIELD_TYPE, RECURSIVE_DEPTH}
 import org.apache.spark.sql.avro.AvroOptions.RECURSIVE_FIELD_MAX_DEPTH_LIMIT
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.types.Decimal.minBytesForPrecision
 
@@ -95,6 +96,33 @@ object SchemaConverters extends Logging {
 
   // The property specifies Catalyst type of the given field
   private val CATALYST_TYPE_PROP_NAME = "spark.sql.catalyst.type"
+  // Avro map keys are always STRING; stamp CHAR/VARCHAR key types on the map schema.
+  private val CATALYST_MAP_KEY_TYPE_PROP_NAME = "spark.sql.catalyst.mapKey.type"
+
+  private def avroStringSchema(catalystType: StringType): Schema = {
+    val stringSchema = SchemaBuilder.builder().stringType()
+    CharVarcharUtils.charVarcharTypeName(catalystType).foreach { name =>
+      stringSchema.addProp(CATALYST_TYPE_PROP_NAME, name)
+    }
+    stringSchema
+  }
+
+  private def avroMapSchema(keyType: StringType, valueSchema: Schema): Schema = {
+    val mapSchema = SchemaBuilder.builder().map().values(valueSchema)
+    CharVarcharUtils.charVarcharTypeName(keyType).foreach { name =>
+      mapSchema.addProp(CATALYST_MAP_KEY_TYPE_PROP_NAME, name)
+    }
+    mapSchema
+  }
+
+  private def parseStampedStringType(catalystTypeAttrValue: String): StringType = {
+    CatalystSqlParser.parseDataType(catalystTypeAttrValue) match {
+      case s: StringType => s
+      case other =>
+        throw new IncompatibleSchemaException(
+          s"Avro $CATALYST_TYPE_PROP_NAME for STRING must be a STRING subtype, got $other")
+    }
+  }
 
   private def toSqlTypeHelper(
       avroSchema: Schema,
@@ -114,7 +142,14 @@ object SchemaConverters extends Logging {
           }
           SchemaType(catalystType, nullable = false)
       }
-      case STRING => SchemaType(StringType, nullable = false)
+      case STRING =>
+        val catalystTypeAttrValue = avroSchema.getProp(CATALYST_TYPE_PROP_NAME)
+        val catalystType = if (catalystTypeAttrValue == null) {
+          StringType
+        } else {
+          parseStampedStringType(catalystTypeAttrValue)
+        }
+        SchemaType(catalystType, nullable = false)
       case BOOLEAN => SchemaType(BooleanType, nullable = false)
       case BYTES | FIXED => avroSchema.getLogicalType match {
         // For FIXED type, if the precision requires more bytes than fixed size, the logical
@@ -244,8 +279,14 @@ object SchemaConverters extends Logging {
           )
           null
         } else {
+          val keyAttr = avroSchema.getProp(CATALYST_MAP_KEY_TYPE_PROP_NAME)
+          val keyType = if (keyAttr == null) {
+            StringType
+          } else {
+            parseStampedStringType(keyAttr)
+          }
           SchemaType(
-            MapType(StringType, schemaType.dataType, valueContainsNull = schemaType.nullable),
+            MapType(keyType, schemaType.dataType, valueContainsNull = schemaType.nullable),
             nullable = false)
         }
 
@@ -369,7 +410,9 @@ object SchemaConverters extends Logging {
 
       case FloatType => builder.floatType()
       case DoubleType => builder.doubleType()
-      case StringType => builder.stringType()
+      // CharType/VarcharType are StringType subclasses, not the StringType singleton.
+      // Stamp spark.sql.catalyst.type so inference restores the length constraint.
+      case s: StringType => avroStringSchema(s)
       case NullType => builder.nullType()
       case d: DecimalType =>
         val avroType = LogicalTypes.decimal(d.precision, d.scale)
@@ -385,9 +428,8 @@ object SchemaConverters extends Logging {
       case ArrayType(et, containsNull) =>
         builder.array()
           .items(toAvroType(et, containsNull, recordName, nameSpace))
-      case MapType(StringType, vt, valueContainsNull) =>
-        builder.map()
-          .values(toAvroType(vt, valueContainsNull, recordName, nameSpace))
+      case MapType(kt: StringType, vt, valueContainsNull) =>
+        avroMapSchema(kt, toAvroType(vt, valueContainsNull, recordName, nameSpace))
       case st: StructType =>
         val childNameSpace = if (nameSpace != "") s"$nameSpace.$recordName" else recordName
         val fieldsAssembler = builder.record(recordName).namespace(nameSpace).fields()
@@ -489,7 +531,7 @@ object SchemaConverters extends Logging {
       case LongType => builder.longType()
       case FloatType => builder.floatType()
       case DoubleType => builder.doubleType()
-      case StringType => builder.stringType()
+      case s: StringType => avroStringSchema(s)
       case NullType => builder.nullType()
       case DateType => LogicalTypes.date().addToSchema(builder.intType())
       case TimestampType => LogicalTypes.timestampMicros().addToSchema(builder.longType())
@@ -522,9 +564,10 @@ object SchemaConverters extends Logging {
         // Make array types nullable
         Schema.createUnion(nullSchema, arraySchema)
 
-      case MapType(StringType, valueType, _) =>
-        val mapSchema = builder.map()
-          .values(toAvroTypeWithDefaults(valueType, recordName = recordName,
+      case MapType(kt: StringType, valueType, _) =>
+        val mapSchema = avroMapSchema(
+          kt,
+          toAvroTypeWithDefaults(valueType, recordName = recordName,
             namespace = namespace, nestingLevel = nestingLevel + 1))
         // Make map types nullable
         Schema.createUnion(nullSchema, mapSchema)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala

@@ -176,8 +176,8 @@ object OrcUtils extends Logging {
       MapType(catalystKeyType, catalystValueType)
     }
 
-    // The Spark query engine has not completely supported CHAR/VARCHAR type yet, and here we
-    // replace the orc CHAR/VARCHAR with STRING type.
+    // Annotate metadata for the legacy path; under charVarcharFirstClassTypes the constrained
+    // types are kept so ORC catalyst attributes / native char/varchar round-trip as first-class.
     CharVarcharUtils.replaceCharVarcharWithStringInSchema(toStructType(schema))
   }
 
@@ -469,9 +469,17 @@ object OrcUtils extends Logging {
           val typeDesc = new TypeDescription(ops.orcCategory)
           typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, dt.typeName)
           Some(typeDesc)
-        case _: StringType =>
+        // Write CHAR/VARCHAR as ORC STRING plus spark.sql.catalyst.type, not native
+        // ORC CHAR/VARCHAR. Native ORC maxLength would truncate/pad independently of
+        // Spark store assignment. Hive-written native CHAR still round-trips on read
+        // via toCatalystSchema. Unbounded STRING (including collated) stamps
+        // StringType.typeName ("string"), matching Avro: this PR does not round-trip
+        // collation on file-only reads.
+        case s: StringType =>
           val typeDesc = new TypeDescription(TypeDescription.Category.STRING)
-          typeDesc.setAttribute(CATALYST_TYPE_ATTRIBUTE_NAME, StringType.typeName)
+          typeDesc.setAttribute(
+            CATALYST_TYPE_ATTRIBUTE_NAME,
+            CharVarcharUtils.charVarcharTypeName(s).getOrElse(StringType.typeName))
           Some(typeDesc)
         case _ => None
       }