[SPARK-58814][SQL] Cover CHAR/VARCHAR round-trips and stop ORC truncating scans

connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala

@@ -40,7 +40,7 @@ import org.apache.spark.sql.TestingUDT.IntervalData
 import org.apache.spark.sql.avro.AvroCompressionCodec._
 import org.apache.spark.sql.catalyst.expressions.AttributeReference
 import org.apache.spark.sql.catalyst.plans.logical.Filter
-import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, DateTimeTestUtils}
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC}
 import org.apache.spark.sql.execution.{FormattedMode, SparkPlan}
 import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition}
@@ -3724,6 +3724,82 @@ abstract class AvroSuite
     }
   }
 
+  test("SPARK-58814: Avro infers nested CHAR/VARCHAR schema and values") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      withTempPath { dir =>
+        val path = dir.getCanonicalPath
+        val input = spark.range(1).selectExpr(
+          "cast('ab' AS CHAR(4)) AS c",
+          "cast('xy' AS VARCHAR(3)) AS v",
+          "named_struct('c', cast('z' AS CHAR(2))) AS s",
+          "array(cast('q' AS VARCHAR(2))) AS a",
+          "map(cast('k' AS CHAR(2)), cast('v' AS VARCHAR(2))) AS m")
+        input.write.mode("overwrite").format("avro").save(path)
+
+        val readBack = spark.read.format("avro").load(path)
+        assert(DataType.equalsIgnoreNullability(readBack.schema, input.schema))
+        checkAnswer(
+          readBack.selectExpr("concat('<', c, '>')", "v", "concat('<', s.c, '>')"),
+          Row("<ab  >", "xy", "<z >"))
+
+        withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+          assert(DataType.equalsIgnoreNullability(
+            spark.read.format("avro").load(path).schema,
+            CharVarcharUtils.replaceCharVarcharWithString(input.schema)))
+        }
+        withSQLConf(
+            SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
+            SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+          assert(DataType.equalsIgnoreNullability(
+            spark.read.format("avro").load(path).schema,
+            input.schema))
+        }
+      }
+
+      withTempPath { dir =>
+        Seq("ab").toDF("c").write.format("avro").save(dir.getCanonicalPath)
+        val charDf = spark.read.schema("c CHAR(4)").format("avro").load(dir.getCanonicalPath)
+        checkAnswer(
+          charDf.selectExpr("concat('<', c, '>')"),
+          Row("<ab  >"))
+      }
+      withTempPath { dir =>
+        Seq("abcdef").toDF("c").write.format("avro").save(dir.getCanonicalPath)
+        Seq("CHAR", "VARCHAR").foreach { typ =>
+          checkError(
+            exception = intercept[SparkRuntimeException] {
+              spark.read.schema(s"c $typ(4)").format("avro")
+                .load(dir.getCanonicalPath).collect()
+            },
+            condition = "EXCEED_LIMIT_LENGTH",
+            parameters = Map("limit" -> "4"))
+        }
+      }
+
+      withTable("avro_char_varchar_assignment") {
+        sql(
+          """CREATE TABLE avro_char_varchar_assignment
+            |(c CHAR(4), v VARCHAR(4)) USING avro""".stripMargin)
+        sql("INSERT INTO avro_char_varchar_assignment VALUES ('ab', 'xy')")
+        assert(spark.table("avro_char_varchar_assignment").schema.map(_.dataType) ===
+          Seq(CharType(4), VarcharType(4)))
+        checkAnswer(
+          sql(
+            """SELECT concat('<', c, '>'), v
+              |FROM avro_char_varchar_assignment""".stripMargin),
+          Row("<ab  >", "xy"))
+        checkError(
+          exception = intercept[SparkRuntimeException] {
+            sql(
+              """INSERT INTO avro_char_varchar_assignment
+                |VALUES ('abcde', 'xy')""".stripMargin).collect()
+          },
+          condition = "EXCEED_LIMIT_LENGTH",
+          parameters = Map("limit" -> "4"))
+      }
+    }
+  }
+
 }
 
 class AvroV1Suite extends AvroSuite {

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

@@ -136,7 +136,7 @@ class OrcDeserializer(
       case DoubleType => (ordinal, value) =>
         updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
 
-      case StringType => (ordinal, value) =>
+      case _: StringType => (ordinal, value) =>
         updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
 
       case BinaryType => (ordinal, value) =>

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

@@ -437,6 +437,11 @@ object OrcUtils extends Logging {
       s"array<${getOrcSchemaString(a.elementType)}>"
     case m: MapType =>
       s"map<${getOrcSchemaString(m.keyType)},${getOrcSchemaString(m.valueType)}>"
+    // Under standard semantics, keep Spark responsible for CHAR/VARCHAR assignment and scan
+    // checks. Native ORC would truncate or pad before Spark can validate the original value.
+    // Preserve-only mode retains the native constrained schema and its legacy enforcement.
+    case _: CharType | _: VarcharType if SQLConf.get.charVarcharStandardSemantics =>
+      StringType.catalogString
     case _: DayTimeIntervalType | _: TimestampNTZType => LongType.catalogString
     case _: YearMonthIntervalType => IntegerType.catalogString
     // Framework types (TimeType, nanosecond timestamps) supply their own ORC schema string.

sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala

@@ -1697,33 +1697,151 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
         sql("DROP TEMPORARY FUNCTION IF EXISTS std_char_param")
         sql("DROP TEMPORARY FUNCTION IF EXISTS std_varchar_param")
       }
+    }
+  }
 
-      // ORC catalog tables stamp the catalyst type so typeof survives write/read.
-      withTable("std_orc") {
-        sql("CREATE TABLE std_orc (c CHAR(5), v VARCHAR(5)) USING orc")
-        sql("INSERT INTO std_orc VALUES ('ab', 'cd')")
-        assert(spark.table("std_orc").schema.map(_.dataType) ===
-          Seq(CharType(5), VarcharType(5)))
-        checkAnswer(
-          sql("SELECT concat('<', c, '>'), concat('<', v, '>') FROM std_orc"),
-          Row("<ab   >", "<cd>"))
+  test("SPARK-58814: major formats preserve CHAR/VARCHAR schemas and values") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      Seq("parquet", "orc").foreach { format =>
+        Seq("v1" -> format, "v2" -> "").foreach { case (sourceVersion, useV1List) =>
+          withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+            withTempPath { dir =>
+              val path = dir.getCanonicalPath
+              val input = spark.range(1).selectExpr(
+                "cast('ab' AS CHAR(4)) AS c",
+                "cast('xy' AS VARCHAR(3)) AS v",
+                "named_struct('c', cast('z' AS CHAR(2))) AS s",
+                "array(cast('q' AS VARCHAR(2))) AS a",
+                "map(cast('k' AS CHAR(2)), cast('v' AS VARCHAR(2))) AS m")
+              input.write.mode("overwrite").format(format).save(path)
+
+              val vectorizedReaderModes = if (format == "orc") Seq(true, false) else Seq(true)
+              vectorizedReaderModes.foreach { vectorizedReaderEnabled =>
+                withSQLConf(
+                    SQLConf.ORC_VECTORIZED_READER_ENABLED.key ->
+                      vectorizedReaderEnabled.toString) {
+                  val readBack = spark.read.format(format).load(path)
+                  assert(DataType.equalsIgnoreNullability(readBack.schema, input.schema),
+                    s"$format $sourceVersion lost CHAR/VARCHAR schema")
+                  checkAnswer(
+                    readBack.selectExpr(
+                      "concat('<', c, '>')",
+                      "v",
+                      "concat('<', s.c, '>')",
+                      "a",
+                      "m"),
+                    Row("<ab  >", "xy", "<z >", Seq("q"), Map("k " -> "v")))
+
+                  withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
+                    val readOff = spark.read.format(format).load(path)
+                    assert(DataType.equalsIgnoreNullability(
+                      readOff.schema,
+                      CharVarcharUtils.replaceCharVarcharWithString(input.schema)))
+                  }
+                  withSQLConf(
+                      SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
+                      SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+                    assert(DataType.equalsIgnoreNullability(
+                      spark.read.format(format).load(path).schema,
+                      input.schema))
+                  }
+                }
+              }
+            }
+          }
+        }
       }
 
-      // File-only ORC inference recovers the catalyst type stamped on write.
-      withTempPath { dir =>
-        val path = dir.getCanonicalPath
-        spark.range(1).selectExpr("cast('ab' AS CHAR(4)) AS c")
-          .write.mode("overwrite").orc(path)
-        val orcDf = spark.read.orc(path)
-        assert(orcDf.schema.head.dataType === CharType(4))
-        checkAnswer(orcDf.selectExpr("concat('<', c, '>')"), Row("<ab  >"))
-        // Reading with first-class types off replaces CHAR with STRING even if the
-        // file was stamped under standardSemantics.
-        withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
-          val readOff = spark.read.orc(path)
-          assert(readOff.schema.head.dataType === StringType)
+      Seq("parquet", "orc", "csv").foreach { format =>
+        Seq("v1" -> format, "v2" -> "").foreach { case (sourceVersion, useV1List) =>
+          withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+            withTempPath { dir =>
+              Seq("ab").toDF("c").write.format(format).save(dir.getCanonicalPath)
+              val charDf = spark.read.schema("c CHAR(4)").format(format)
+                .load(dir.getCanonicalPath)
+              checkAnswer(charDf.selectExpr("concat('<', c, '>')"), Row("<ab  >"))
+            }
+            withTempPath { dir =>
+              Seq("abcdef").toDF("c").write.format(format).save(dir.getCanonicalPath)
+              Seq("CHAR", "VARCHAR").foreach { typ =>
+                withClue(s"$format $sourceVersion $typ: ") {
+                  val readDf = spark.read.schema(s"c $typ(4)").format(format)
+                    .load(dir.getCanonicalPath)
+                  checkError(
+                    exception = intercept[SparkRuntimeException] {
+                      readDf.collect()
+                    },
+                    condition = "EXCEED_LIMIT_LENGTH",
+                    parameters = Map("limit" -> "4"))
+                }
+              }
+            }
+
+            val table = s"std_${format}_${sourceVersion}_assignment"
+            withTable(table) {
+              sql(s"CREATE TABLE $table (c CHAR(4), v VARCHAR(4)) USING $format")
+              sql(s"INSERT INTO $table VALUES ('ab', 'xy')")
+              assert(spark.table(table).schema.map(_.dataType) ===
+                Seq(CharType(4), VarcharType(4)))
+              checkAnswer(
+                sql(s"SELECT concat('<', c, '>'), v FROM $table"),
+                Row("<ab  >", "xy"))
+              checkError(
+                exception = intercept[SparkRuntimeException] {
+                  sql(s"INSERT INTO $table VALUES ('abcde', 'xy')").collect()
+                },
+                condition = "EXCEED_LIMIT_LENGTH",
+                parameters = Map("limit" -> "4"))
+            }
+          }
         }
       }
+
+      Seq("v1" -> "orc", "v2" -> "").foreach { case (sourceVersion, useV1List) =>
+        Seq(true, false).foreach { vectorizedReaderEnabled =>
+          withSQLConf(
+              SQLConf.USE_V1_SOURCE_LIST.key -> useV1List,
+              SQLConf.ORC_VECTORIZED_READER_ENABLED.key -> vectorizedReaderEnabled.toString) {
+            withTempPath { dir =>
+              val path = dir.getCanonicalPath
+              spark.range(1).selectExpr(
+                "named_struct('c', 'abcdef') AS s",
+                "array('abcdef') AS a",
+                "map('abcdef', 'ok') AS mk",
+                "map('ok', 'abcdef') AS mv")
+                .write.mode("overwrite").orc(path)
+              val readBack = spark.read.schema(
+                """s STRUCT<c: CHAR(4)>,
+                  |a ARRAY<VARCHAR(4)>,
+                  |mk MAP<CHAR(4), VARCHAR(4)>,
+                  |mv MAP<CHAR(4), VARCHAR(4)>""".stripMargin).orc(path)
+              Seq("s.c", "a", "mk", "mv").foreach { field =>
+                withClue(
+                    s"ORC $sourceVersion vectorized=$vectorizedReaderEnabled $field: ") {
+                  checkError(
+                    exception = intercept[SparkRuntimeException] {
+                      readBack.selectExpr(field).collect()
+                    },
+                    condition = "EXCEED_LIMIT_LENGTH",
+                    parameters = Map("limit" -> "4"))
+                }
+              }
+            }
+            withSQLConf(
+                SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
+                SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
+              withTempPath { dir =>
+                val path = dir.getCanonicalPath
+                Seq("abcdef").toDF("v").write.mode("overwrite").orc(path)
+                val readBack = spark.read.schema("v VARCHAR(4)").orc(path)
+                assert(readBack.schema.head.dataType === VarcharType(4))
+                checkAnswer(readBack, Row("abcd"))
+              }
+            }
+          }
+        }
+      }
+
       // First-class types off: CAST CHAR is STRING before the writer, so ORC does not stamp CHAR.
       withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false") {
         withTempPath { dir =>
@@ -1733,17 +1851,6 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
           assert(spark.read.orc(path).schema.head.dataType === StringType)
         }
       }
-      // preserveCharVarcharTypeInfo also keeps first-class types, so write still stamps CHAR.
-      withSQLConf(
-          SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "false",
-          SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key -> "true") {
-        withTempPath { dir =>
-          val path = dir.getCanonicalPath
-          spark.range(1).selectExpr("cast('ab' AS CHAR(4)) AS c")
-            .write.mode("overwrite").orc(path)
-          assert(spark.read.orc(path).schema.head.dataType === CharType(4))
-        }
-      }
       // ORC stamps collated unbounded STRING as plain "string"; the inferred type is the
       // same as Avro, which omits the catalyst property on unbounded STRING.
       withTempPath { dir =>
@@ -1825,21 +1932,14 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
         }
       }
 
-      // JSON / CSV keep a user-specified CHAR/VARCHAR schema under the flag.
+      // JSON has no embedded schema, so a user-specified schema supplies the logical type.
       withTempPath { dir =>
         val path = dir.getCanonicalPath
         spark.range(1).selectExpr("cast(id AS STRING) AS c").write.mode("overwrite")
           .json(s"$path/json")
         val jsonDf = spark.read.schema("c CHAR(5)").json(s"$path/json")
         assert(jsonDf.schema.head.dataType === CharType(5))
         checkAnswer(jsonDf.selectExpr("concat('<', c, '>')"), Row("<0    >"))
-
-        spark.range(1).selectExpr("cast(id AS STRING) AS c").write.mode("overwrite")
-          .option("header", "true").csv(s"$path/csv")
-        val csvDf = spark.read.schema("c VARCHAR(5)").option("header", "true")
-          .csv(s"$path/csv")
-        assert(csvDf.schema.head.dataType === VarcharType(5))
-        checkAnswer(csvDf, Row("0"))
       }
     }
   }