[SPARK-59274][SQL] Enforce CHAR/VARCHAR semantics in schema-driven text parsing

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

@@ -278,8 +278,10 @@ class UnivocityParser(
         timeFormatter.parse(datum)
       }
 
-    case _: StringType => (d: String) =>
-      nullSafeDatum(d, name, nullable, options)(UTF8String.fromString)
+    case dt: StringType => (d: String) =>
+      nullSafeDatum(d, name, nullable, options) { s =>
+        CharVarcharUtils.applyTextParseSemantics(UTF8String.fromString(s), dt)
+      }
 
     case _: BinaryType => (d: String) =>
       nullSafeDatum(d, name, nullable, options)(_.getBytes)

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

@@ -183,7 +183,7 @@ case class SchemaOfCsv(
       DataTypeMismatch(
         errorSubClass = "UNEXPECTED_NULL",
         messageParameters = Map("exprName" -> "csv"))
-    } else if (child.dataType != StringType) {
+    } else if (!child.dataType.isInstanceOf[StringType]) {
       DataTypeMismatch(
         errorSubClass = "UNEXPECTED_INPUT_TYPE",
         messageParameters = Map(

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

@@ -1913,7 +1913,7 @@ case class SchemaOfJson(
       DataTypeMismatch(
         errorSubClass = "UNEXPECTED_NULL",
         messageParameters = Map("exprName" -> "json"))
-    } else if (child.dataType != StringType) {
+    } else if (!child.dataType.isInstanceOf[StringType]) {
       DataTypeMismatch(
         errorSubClass = "UNEXPECTED_INPUT_TYPE",
         messageParameters = Map(

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

@@ -200,7 +200,7 @@ case class SchemaOfXml(
       DataTypeMismatch(
         errorSubClass = "UNEXPECTED_NULL",
         messageParameters = Map("exprName" -> "xml"))
-    } else if (child.dataType != StringType) {
+    } else if (!child.dataType.isInstanceOf[StringType]) {
       DataTypeMismatch(
         errorSubClass = "UNEXPECTED_INPUT_TYPE",
         messageParameters = Map(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala

@@ -27,7 +27,7 @@ import scala.util.control.NonFatal
 import com.fasterxml.jackson.core._
 import org.apache.hadoop.fs.PositionedReadable
 
-import org.apache.spark.SparkUpgradeException
+import org.apache.spark.{SparkRuntimeException, SparkUpgradeException}
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, StructFilters}
 import org.apache.spark.sql.catalyst.expressions._
@@ -40,7 +40,7 @@ import org.apache.spark.sql.sources.Filter
 import org.apache.spark.sql.types._
 import org.apache.spark.types.variant._
 import org.apache.spark.unsafe.types.{CalendarInterval, TimestampNanosVal, UTF8String, VariantVal}
-import org.apache.spark.util.Utils
+import org.apache.spark.util.{SparkErrorUtils, Utils}
 
 /**
  * Constructs a parser for a given schema that translates a json string to an [[InternalRow]].
@@ -54,6 +54,16 @@ class JacksonParser(
   import JacksonUtils._
   import com.fasterxml.jackson.core.JsonToken._
 
+  private object DuplicateMapKeyException {
+    def unapply(exception: Throwable): Option[SparkRuntimeException] = {
+      SparkErrorUtils.getRootCause(exception) match {
+        case cause: SparkRuntimeException if cause.getCondition == "DUPLICATED_MAP_KEY" =>
+          Some(cause)
+        case _ => None
+      }
+    }
+  }
+
   // A `ValueConverter` is responsible for converting a value from `JsonParser`
   // to a value in a field for `InternalRow`.
   private type ValueConverter = JsonParser => AnyRef
@@ -187,7 +197,8 @@ class JacksonParser(
   private def makeMapRootConverter(mt: MapType): JsonParser => Iterable[InternalRow] = {
     val fieldConverter = makeConverter(mt.valueType)
     (parser: JsonParser) => parseJsonToken[Iterable[InternalRow]](parser, mt) {
-      case START_OBJECT => Some(InternalRow(convertMap(parser, fieldConverter)))
+      case START_OBJECT =>
+        Some(InternalRow(convertMap(parser, fieldConverter, mt.keyType, mt.valueType)))
     }
   }
 
@@ -302,7 +313,7 @@ class JacksonParser(
           }
       }
 
-    case _: StringType => (parser: JsonParser) => {
+    case dt: StringType => (parser: JsonParser) => {
       // This must be enabled if we will retrieve the bytes directly from the raw content:
       val oldFeature = parser.getFeatureMask
       val featureToAdd = JsonParser.Feature.INCLUDE_SOURCE_IN_LOCATION.getMask
@@ -356,7 +367,7 @@ class JacksonParser(
       // to be reset. This ensures that every feature is restored to its previous
       // state as defined by `oldFeature`.
       parser.overrideStdFeatures(oldFeature, ~0)
-      result
+      CharVarcharUtils.applyTextParseSemantics(result, dt)
     }
 
     case TimestampType =>
@@ -480,7 +491,7 @@ class JacksonParser(
     case mt: MapType =>
       val valueConverter = makeConverter(mt.valueType)
       (parser: JsonParser) => parseJsonToken[MapData](parser, dataType) {
-        case START_OBJECT => convertMap(parser, valueConverter)
+        case START_OBJECT => convertMap(parser, valueConverter, mt.keyType, mt.valueType)
       }
 
     case udt: UserDefinedType[_] =>
@@ -589,6 +600,7 @@ class JacksonParser(
             bitmask(index) = false
           } catch {
             case e: SparkUpgradeException => throw e
+            case DuplicateMapKeyException(e) => throw e
             case err: PartialValueException if enablePartialResults =>
               badRecordException = badRecordException.orElse(Some(err.cause))
               row.update(index, err.partialResult)
@@ -617,33 +629,58 @@ class JacksonParser(
    */
   private def convertMap(
       parser: JsonParser,
-      fieldConverter: ValueConverter): MapData = {
+      fieldConverter: ValueConverter,
+      keyType: DataType,
+      valueType: DataType): MapData = {
     val keys = ArrayBuffer.empty[UTF8String]
     val values = ArrayBuffer.empty[Any]
-    var badRecordException: Option[Throwable] = None
+    var partialResultException: Option[Throwable] = None
+    var badMapException: Option[Throwable] = None
 
     while (nextUntil(parser, JsonToken.END_OBJECT)) {
-      keys += UTF8String.fromString(parser.currentName)
-      try {
-        values += fieldConverter.apply(parser)
+      val rawKey = UTF8String.fromString(parser.currentName)
+      val value = try {
+        Some(fieldConverter.apply(parser))
       } catch {
         case err: PartialValueException if enablePartialResults =>
-          badRecordException = badRecordException.orElse(Some(err.cause))
-          values += err.partialResult
+          partialResultException = partialResultException.orElse(Some(err.cause))
+          Some(err.partialResult)
+        case DuplicateMapKeyException(e) => throw e
         case NonFatal(e) if enablePartialResults =>
-          badRecordException = badRecordException.orElse(Some(e))
+          badMapException = badMapException.orElse(Some(e))
           parser.skipChildren()
+          None
+      }
+      value.foreach { parsedValue =>
+        try {
+          val key = CharVarcharUtils.applyTextParseSemantics(rawKey, keyType)
+          keys += key
+          values += parsedValue
+        } catch {
+          case DuplicateMapKeyException(e) => throw e
+          case NonFatal(e) if enablePartialResults =>
+            badMapException = badMapException.orElse(Some(e))
+        }
       }
     }
 
-    // The JSON map will never have null or duplicated map keys, it's safe to create a
-    // ArrayBasedMapData directly here.
-    val mapData = ArrayBasedMapData(keys.toArray, values.toArray)
+    val mapData = keyType match {
+      case _: CharType | _: VarcharType =>
+        new ArrayBasedMapBuilder(keyType, valueType).from(
+          new GenericArrayData(keys.toArray), new GenericArrayData(values.toArray))
+      case _ =>
+        // Preserve the historical behavior for ordinary string keys.
+        ArrayBasedMapData(keys.toArray, values.toArray)
+    }
 
-    if (badRecordException.isEmpty) {
+    // Ordinary value or key conversion failures invalidate the whole map. Delay throwing until
+    // the closing brace has been consumed and constrained-key deduplication has been applied.
+    badMapException.foreach(throw _)
+
+    if (partialResultException.isEmpty) {
       mapData
     } else {
-      throw PartialMapDataResultException(mapData, badRecordException.get)
+      throw PartialMapDataResultException(mapData, partialResultException.get)
     }
   }
 
@@ -717,6 +754,7 @@ class JacksonParser(
       }
     } catch {
       case e: SparkUpgradeException => throw e
+      case DuplicateMapKeyException(e) => throw e
       case e @ (_: RuntimeException | _: JsonProcessingException | _: MalformedInputException) =>
         // JSON parser currently doesn't support partial results for corrupted records.
         // For such records, all fields other than the field configured by

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CharVarcharUtils.scala

@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
 import org.apache.spark.util.ArrayImplicits._
 
 object CharVarcharUtils extends Logging with SparkCharVarcharUtils {
@@ -156,6 +157,25 @@ object CharVarcharUtils extends Logging with SparkCharVarcharUtils {
     StructType(fields)
   }
 
+  /**
+   * Applies CHAR padding and VARCHAR length checks when parsing text into a typed schema.
+   * Null stays null. Unbounded STRING is unchanged. This is assignment semantics
+   * (overflow raises EXCEED_LIMIT_LENGTH), not explicit CAST truncation.
+   */
+  def applyTextParseSemantics(value: UTF8String, dt: DataType): UTF8String = {
+    if (value == null) {
+      null
+    } else {
+      dt match {
+        case c: CharType =>
+          CharVarcharCodegenUtils.charTypeWriteSideCheck(value, c.length)
+        case v: VarcharType =>
+          CharVarcharCodegenUtils.varcharTypeWriteSideCheck(value, v.length)
+        case _ => value
+      }
+    }
+  }
+
   /**
    * Returns an expression to apply write-side string length check for the given expression. A
    * string value can not exceed N characters if it's written into a CHAR(N)/VARCHAR(N)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/FailureSafeParser.scala

@@ -17,11 +17,13 @@
 
 package org.apache.spark.sql.catalyst.util
 
+import org.apache.spark.SparkRuntimeException
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
 import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.util.SparkErrorUtils
 
 class FailureSafeParser[IN](
     rawParser: IN => Iterable[InternalRow],
@@ -34,6 +36,13 @@ class FailureSafeParser[IN](
   private val resultRow = new GenericInternalRow(schema.length)
   private val nullResult = new GenericInternalRow(schema.length)
 
+  private def duplicateMapKeyCause(e: Throwable): Option[SparkRuntimeException] = {
+    SparkErrorUtils.getRootCause(e) match {
+      case cause: SparkRuntimeException if cause.getCondition == "DUPLICATED_MAP_KEY" => Some(cause)
+      case _ => None
+    }
+  }
+
   // This function takes 2 parameters: an optional partial result, and the bad record. If the given
   // schema doesn't contain a field for corrupted record, we just return the partial result or a
   // row with all fields null. If the given schema contains a field for corrupted record, we will
@@ -59,6 +68,9 @@ class FailureSafeParser[IN](
     try {
       rawParser.apply(input).iterator.map(row => toResultRow(Some(row), () => null))
     } catch {
+      // Duplicate map keys are governed by mapKeyDedupPolicy, not the parse mode.
+      case e: BadRecordException if duplicateMapKeyCause(e).isDefined =>
+        throw duplicateMapKeyCause(e).get
       case e: BadRecordException => mode match {
         case PermissiveMode =>
           val partialResults = e.partialResults()