[SPARK-59273][SQL] Complete CHAR/VARCHAR support at core execution boundaries

sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameNaFunctions.scala

@@ -249,7 +249,7 @@ final class DataFrameNaFunctions private[sql](df: DataFrame)
     val projections = outputAttributes.map { col =>
       val typeMatches = (targetType, col.dataType) match {
         case (NumericType, dt) => dt.isInstanceOf[NumericType]
-        case (StringType, dt) => dt == StringType
+        case (StringType, _: StringType) => true
         case (BooleanType, dt) => dt == BooleanType
         case _ =>
           throw new IllegalArgumentException(s"$targetType is not matched at fillValue")

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

@@ -298,7 +298,7 @@ private object RowToColumnConverter {
       case LongType | TimestampType | TimestampNTZType | _: DayTimeIntervalType | _: TimeType =>
         LongConverter
       case DoubleType => DoubleConverter
-      case StringType => StringConverter
+      case _: StringType => StringConverter
       case _: GeographyType | _: GeometryType => BinaryViewConverter
       case CalendarIntervalType => CalendarConverter
       case VariantType => VariantConverter

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala

@@ -142,7 +142,6 @@ case class AnalyzeColumnCommand(
     case DoubleType | FloatType => true
     case BooleanType => true
     case _: DatetimeType => true
-    case _: CharType | _: VarcharType => false
     case BinaryType | _: StringType => true
     case _ => false
   }

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

@@ -550,7 +550,7 @@ object PartitioningUtils extends SQLConfHelper {
       zoneId: ZoneId): Any = desiredType match {
     case _ if value == DEFAULT_PARTITION_NAME => null
     case NullType => null
-    case StringType => UTF8String.fromString(unescapePathName(value))
+    case _: StringType => UTF8String.fromString(unescapePathName(value))
     case ByteType => Integer.parseInt(value).toByte
     case ShortType => Integer.parseInt(value).toShort
     case IntegerType => Integer.parseInt(value)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala

@@ -296,7 +296,7 @@ private[jdbc] object JDBCValueGetter {
           (t: Timestamp) => localDateTimeToMicros(dialect.convertJavaTimestampToTimestampNTZ(t))
         }
 
-      case StringType =>
+      case _: StringType =>
         arrayConverter[Object]((obj: Object) => UTF8String.fromString(obj.toString))
 
       case DateType => arrayConverter[Date] {

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -213,7 +213,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
     case java.sql.Types.BIT => BooleanType // @see JdbcDialect for quirks
     case java.sql.Types.BLOB => BinaryType
     case java.sql.Types.BOOLEAN => BooleanType
-    case java.sql.Types.CHAR if conf.charVarcharAsString => StringType
+    case java.sql.Types.CHAR
+        if conf.charVarcharAsString && !conf.charVarcharFirstClassTypes => StringType
     case java.sql.Types.CHAR => CharType(precision)
     case java.sql.Types.CLOB => StringType
     case java.sql.Types.DATE => DateType
@@ -261,7 +262,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
       } else getTimestampType(isTimestampNTZ)
     case java.sql.Types.TINYINT => IntegerType
     case java.sql.Types.VARBINARY => BinaryType
-    case java.sql.Types.VARCHAR if conf.charVarcharAsString => StringType
+    case java.sql.Types.VARCHAR
+        if conf.charVarcharAsString && !conf.charVarcharFirstClassTypes => StringType
     case java.sql.Types.VARCHAR => VarcharType(precision)
     case java.sql.Types.NULL => NullType
     case _ =>
@@ -469,8 +471,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
     case LongType => JDBCValueGetter.LongGetter
     case ShortType => JDBCValueGetter.ShortGetter
     case ByteType => JDBCValueGetter.ByteGetter
-    case StringType if metadata.contains("rowid") => JDBCValueGetter.RowIdGetter
-    case StringType => JDBCValueGetter.StringGetter
+    case _: StringType if metadata.contains("rowid") => JDBCValueGetter.RowIdGetter
+    case _: StringType => JDBCValueGetter.StringGetter
     case TimestampType if metadata.contains("logical_time_type") =>
       JDBCValueGetter.LogicalTimeGetter
     case TimestampType => JDBCValueGetter.TimestampGetter(dialect)
@@ -530,7 +532,7 @@ object JdbcUtils extends Logging with SQLConfHelper {
       (stmt: PreparedStatement, row: Row, pos: Int) =>
         stmt.setBoolean(pos + 1, row.getBoolean(pos))
 
-    case StringType =>
+    case _: StringType =>
       (stmt: PreparedStatement, row: Row, pos: Int) =>
         stmt.setString(pos + 1, row.getString(pos))
 

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

@@ -22,7 +22,7 @@ import scala.jdk.CollectionConverters._
 import org.apache.spark.SparkUnsupportedOperationException
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{StringType, StructType}
+import org.apache.spark.sql.types.{CharType, StringType, StructField, StructType, VarcharType}
 
 class DataFrameNaFunctionsSuite extends SharedSparkSession {
   import testImplicits._
@@ -215,6 +215,17 @@ class DataFrameNaFunctionsSuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-59273: fill CHAR/VARCHAR columns") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      val schema = StructType(Seq(
+        StructField("c", CharType(3)),
+        StructField("v", VarcharType(3))))
+      val input = spark.createDataFrame(sparkContext.parallelize(Seq(Row(null, null))), schema)
+
+      checkAnswer(input.na.fill("x"), Row("x  ", "x"))
+    }
+  }
+
   test("fill with map") {
     withSQLConf(SQLConf.SUPPORT_QUOTED_REGEX_COLUMN_NAME.key -> "false") {
       val df = Seq[(String, String, java.lang.Integer, java.lang.Long,

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

@@ -158,6 +158,24 @@ class StatisticsCollectionSuite extends StatisticsCollectionTestBase with Shared
     }
   }
 
+  test("SPARK-59273: collect CHAR/VARCHAR column statistics") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      val tableName = "char_varchar_column_stats"
+      withTable(tableName) {
+        sql(s"CREATE TABLE $tableName(c CHAR(3), v VARCHAR(3)) USING parquet")
+        sql(s"INSERT INTO $tableName VALUES ('a', 'x'), ('bb', 'yz'), (NULL, NULL)")
+        sql(s"ANALYZE TABLE $tableName COMPUTE STATISTICS FOR COLUMNS c, v")
+
+        val columnStats = getCatalogTable(tableName).stats.get.colStats
+        assert(columnStats.keySet === Set("c", "v"))
+        assert(columnStats("c").distinctCount.contains(BigInt(2)))
+        assert(columnStats("v").distinctCount.contains(BigInt(2)))
+        assert(columnStats("c").nullCount.contains(BigInt(1)))
+        assert(columnStats("v").nullCount.contains(BigInt(1)))
+      }
+    }
+  }
+
   test("test table-level statistics for data source table") {
     val tableName = "tbl"
     withTable(tableName) {

sql/core/src/test/scala/org/apache/spark/sql/execution/RowToColumnConverterSuite.scala

@@ -107,6 +107,22 @@ class RowToColumnConverterSuite extends SparkFunSuite {
     }
   }
 
+  test("SPARK-59273: CHAR/VARCHAR columns") {
+    val schema = StructType(Seq(
+      StructField("c", CharType(3)),
+      StructField("v", VarcharType(3)),
+      StructField("a", ArrayType(CharType(3)))))
+    val rows = Seq(InternalRow(
+      UTF8String.fromString("a  "),
+      UTF8String.fromString("bc"),
+      new GenericArrayData(Seq(UTF8String.fromString("d  ")))))
+    val vectors = convertRows(rows, schema)
+
+    assert(vectors(0).getUTF8String(0).toString === "a  ")
+    assert(vectors(1).getUTF8String(0).toString === "bc")
+    assert(vectors(2).getArray(0).getUTF8String(0).toString === "d  ")
+  }
+
   test("non-nullable map column with null values") {
     val mapType = MapType(IntegerType, StringType, valueContainsNull = true)
     val schema = StructType(Seq(StructField("m", mapType, nullable = false)))

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala

@@ -1167,6 +1167,34 @@ abstract class ParquetPartitionDiscoverySuite
     }
   }
 
+  test("SPARK-59273: read CHAR/VARCHAR partition values") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      Seq("CHAR(3)" -> "a  ", "VARCHAR(3)" -> "a").foreach { case (dataType, expected) =>
+        withTempPath { path =>
+          Seq((1, "a")).toDF("id", "part")
+            .write.partitionBy("part").parquet(path.getCanonicalPath)
+          val readback = spark.read
+            .schema(s"id INT, part $dataType")
+            .parquet(path.getCanonicalPath)
+
+          checkAnswer(readback, Row(1, expected))
+        }
+      }
+      withTempPath { path =>
+        Seq((1, "abcdef")).toDF("id", "part")
+          .write.partitionBy("part").parquet(path.getCanonicalPath)
+        val readback = spark.read
+          .schema("id INT, part VARCHAR(3)")
+          .parquet(path.getCanonicalPath)
+
+        checkError(
+          exception = intercept[SparkRuntimeException](readback.collect()),
+          condition = "EXCEED_LIMIT_LENGTH",
+          parameters = Map("limit" -> "3"))
+      }
+    }
+  }
+
   test("SPARK-40212: SparkSQL castPartValue does not properly handle byte, short, float") {
     withTempDir { dir =>
       val data = Seq[(Int, Byte, Short, Float)](

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -1820,6 +1820,73 @@ class JDBCSuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-59273: read CHAR/VARCHAR values and arrays") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      val charArray = mock(classOf[java.sql.Array])
+      when(charArray.getArray).thenReturn(Array[AnyRef]("c", "dd"))
+      val varcharArray = mock(classOf[java.sql.Array])
+      when(varcharArray.getArray).thenReturn(Array[AnyRef]("e", "ff"))
+      val rs = mock(classOf[ResultSet])
+      when(rs.next()).thenReturn(true, false)
+      when(rs.getString(1)).thenReturn("a  ")
+      when(rs.getString(2)).thenReturn("bb")
+      when(rs.getArray(3)).thenReturn(charArray)
+      when(rs.getArray(4)).thenReturn(varcharArray)
+      val schema = StructType(Seq(
+        StructField("c", CharType(3)),
+        StructField("v", VarcharType(3)),
+        StructField("ca", ArrayType(CharType(2))),
+        StructField("va", ArrayType(VarcharType(2)))))
+
+      val rows = JdbcUtils.resultSetToSparkInternalRows(
+        rs, NoopDialect, schema, new InputMetrics).toArray
+      assert(rows.length === 1)
+      assert(rows.head.getUTF8String(0).toString === "a  ")
+      assert(rows.head.getUTF8String(1).toString === "bb")
+      assert(rows.head.getArray(2).toObjectArray(CharType(2)).map(_.toString).toSeq ===
+        Seq("c", "dd"))
+      assert(rows.head.getArray(3).toObjectArray(VarcharType(2)).map(_.toString).toSeq ===
+        Seq("e", "ff"))
+    }
+  }
+
+  test("SPARK-59273: first-class modes take precedence in JDBC schema inference") {
+    Seq(
+      SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key,
+      SQLConf.PRESERVE_CHAR_VARCHAR_TYPE_INFO.key).foreach { firstClassConfig =>
+      withSQLConf(
+          firstClassConfig -> "true",
+          SQLConf.LEGACY_CHAR_VARCHAR_AS_STRING.key -> "true") {
+        val df = spark.read.format("jdbc")
+          .option("url", urlWithUserAndPass)
+          .option("dbtable", "TEST.STRTYPES")
+          .load()
+
+        assert(df.schema("B").dataType === VarcharType(20))
+        assert(df.schema("D").dataType === CharType(20))
+        checkAnswer(df.select("B", "D"), Row("Sensitive", "Twenty-byte CHAR    "))
+      }
+    }
+  }
+
+  test("SPARK-59273: write CHAR/VARCHAR values") {
+    withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+      val tableName = "char_varchar_write"
+      sql("SELECT CAST('a' AS CHAR(3)) AS c, CAST('bb' AS VARCHAR(3)) AS v")
+        .write.format("jdbc")
+        .mode("overwrite")
+        .option("url", urlWithUserAndPass)
+        .option("dbtable", tableName)
+        .save()
+
+      val rs = conn.createStatement().executeQuery(s"""SELECT "c", "v" FROM $tableName""")
+      assert(rs.next())
+      assert(rs.getString(1) === "a  ")
+      assert(rs.getString(2) === "bb")
+      rs.close()
+    }
+  }
+
   test("SPARK-58876: Oracle compileValue renders a LocalDateTime as a JDBC timestamp literal") {
     // Filters on an NTZ-mapped Oracle column push down a LocalDateTime; it must become a valid
     // Oracle literal rather than LocalDateTime.toString.