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

### What changes were proposed in this pull request?

When `spark.sql.charVarchar.standardSemantics.enabled` is true, `CharType` and `VarcharType` are first-class `StringType` subtypes. Several execution-boundary matchers still used exact `StringType` (or an explicit CHAR/VARCHAR reject), so those paths failed or skipped constrained string columns.

This patch treats CHAR/VARCHAR as the string family at:

- JDBC getters/setters and JDBC array element conversion
- JDBC schema inference (`CHAR`/`VARCHAR` keep first-class types when standard semantics is on, even if `charVarcharAsString` is also set)
- File partition value decoding
- `RowToColumnConverter`
- `DataFrame.na.fill` for string replacement values
- `ANALYZE TABLE ... FOR COLUMNS` (string-family stats)

Read-side CHAR padding and VARCHAR overflow still come from existing CAST / `ApplyCharTypePadding` paths rather than being reimplemented in each converter.

### Why are the changes needed?

With first-class CHAR/VARCHAR, JDBC scans/writes, file-only partition discovery, columnar conversion, `na.fill("...")`, and column stats currently throw or silently ignore those columns. That blocks enabling standard semantics.

JIRA: https://issues.apache.org/jira/browse/SPARK-59273 (subtask of SPARK-58794)

### Does this PR introduce _any_ user-facing change?

Yes, when `spark.sql.charVarchar.standardSemantics.enabled` is true (still default false):

- JDBC read/write of CHAR/VARCHAR (including arrays) no longer fails with an unsupported JDBC type.
- File partition columns declared as CHAR/VARCHAR can be decoded; CHAR is padded on scan and oversize VARCHAR fails with `EXCEED_LIMIT_LENGTH`.
- Columnar row-to-column conversion accepts CHAR/VARCHAR.
- `df.na.fill("x")` fills null CHAR/VARCHAR columns (CHAR values are padded by CAST).
- `ANALYZE TABLE ... FOR COLUMNS` collects string-family stats on CHAR/VARCHAR instead of rejecting them.

### How was this patch tested?

Added/extended unit tests:

- `JDBCSuite`: read CHAR/VARCHAR and arrays; write CHAR/VARCHAR; standard semantics wins over `charVarcharAsString` in schema inference
- `ParquetV1PartitionDiscoverySuite` / `ParquetV2PartitionDiscoverySuite`: CHAR/VARCHAR partition values and oversize VARCHAR
- `RowToColumnConverterSuite`: CHAR/VARCHAR and nested CHAR arrays
- `DataFrameNaFunctionsSuite`: `na.fill` on CHAR/VARCHAR
- `StatisticsCollectionSuite`: `ANALYZE TABLE ... FOR COLUMNS` on CHAR/VARCHAR

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Cursor Grok 4.6

Closes #58541 from srielau/serge-rielau_data/SPARK-59273.

Authored-by: Serge Rielau <serge@rielau.com>
Signed-off-by: Daniel Tenedorio <daniel.tenedorio@databricks.com>

Author: srielau

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala

@@ -107,7 +107,7 @@ object EstimationUtils {
     8 + attributes.map { attr =>
       if (attrStats.get(attr).map(_.avgLen.isDefined).getOrElse(false)) {
         attr.dataType match {
-          case StringType =>
+          case _: StringType =>
             // UTF8String: base + offset + numBytes
             attrStats(attr).avgLen.get + 8 + 4
           case _ =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala

@@ -293,7 +293,7 @@ case class FilterEstimation(plan: Filter) extends Logging {
     attr.dataType match {
       case _: NumericType | DateType | TimestampType | BooleanType =>
         evaluateBinaryForNumeric(op, attr, literal, update)
-      case StringType | BinaryType =>
+      case _: StringType | BinaryType =>
         // TODO: It is difficult to support other binary comparisons for String/Binary
         // type without min/max and advanced statistics like histogram.
         logDebug("[CBO] No range comparison statistics for String/Binary type " + attr)
@@ -329,7 +329,7 @@ case class FilterEstimation(plan: Filter) extends Logging {
     // We currently don't store min/max for binary/string type. For other types, if min/max are
     // missing, treat the range as unknown (instead of "all nulls") and fall back to NDV/histogram.
     val valueInRange = attr.dataType match {
-      case StringType | BinaryType =>
+      case _: StringType | BinaryType =>
         true
       case _ if !colStat.hasMinMaxStats =>
         true
@@ -353,7 +353,7 @@ case class FilterEstimation(plan: Filter) extends Logging {
       // We update ColumnStat structure after apply this equality predicate:
       // Set distinctCount to 1, nullCount to 0, and min/max values (if exist) to the literal value.
       val newStats = attr.dataType match {
-        case StringType | BinaryType =>
+        case _: StringType | BinaryType =>
           colStat.copy(distinctCount = Some(1), nullCount = Some(0))
         case _ =>
           colStat.copy(distinctCount = Some(1), min = Some(literal.value),
@@ -443,7 +443,7 @@ case class FilterEstimation(plan: Filter) extends Logging {
         }
 
       // We assume the whole set since there is no min/max information for String/Binary type
-      case StringType | BinaryType =>
+      case _: StringType | BinaryType =>
         if (ndv.toDouble == 0)  {
           return Some(0.0)
         }
@@ -686,7 +686,7 @@ case class FilterEstimation(plan: Filter) extends Logging {
     }
 
     attrLeft.dataType match {
-      case StringType | BinaryType =>
+      case _: StringType | BinaryType =>
         // TODO: It is difficult to support other binary comparisons for String/Binary
         // type without min/max and advanced statistics like histogram.
         logDebug("[CBO] No range comparison statistics for String/Binary type " + attrLeft)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/ValueInterval.scala

@@ -53,7 +53,7 @@ object ValueInterval {
       min: Option[Any],
       max: Option[Any],
       dataType: DataType): ValueInterval = dataType match {
-    case StringType | BinaryType => new DefaultValueInterval()
+    case _: StringType | BinaryType => new DefaultValueInterval()
     case _ if min.isEmpty || max.isEmpty => new NullValueInterval()
     case _ =>
       NumericValueInterval(

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/FilterEstimationSuite.scala

@@ -73,6 +73,13 @@ class FilterEstimationSuite extends StatsEstimationTestBase {
   val colStatString = ColumnStat(distinctCount = Some(10), min = None, max = None,
     nullCount = Some(0), avgLen = Some(2), maxLen = Some(2))
 
+  val attrChar = AttributeReference("cchar", CharType(2))()
+  val colStatChar = ColumnStat(distinctCount = Some(10), min = None, max = None,
+    nullCount = Some(0), avgLen = Some(2), maxLen = Some(2))
+  val attrVarchar = AttributeReference("cvarchar", VarcharType(2))()
+  val colStatVarchar = ColumnStat(distinctCount = Some(10), min = None, max = None,
+    nullCount = Some(0), avgLen = Some(2), maxLen = Some(2))
+
   // column cint2 has values: 7, 8, 9, 10, 11, 12, 13, 14, 15, 16
   // Hence, distinctCount:10, min:7, max:16, nullCount:0, avgLen:4, maxLen:4
   // This column is created to test "cint < cint2
@@ -121,6 +128,8 @@ class FilterEstimationSuite extends StatsEstimationTestBase {
     attrDecimal -> colStatDecimal,
     attrDouble -> colStatDouble,
     attrString -> colStatString,
+    attrChar -> colStatChar,
+    attrVarchar -> colStatVarchar,
     attrInt2 -> colStatInt2,
     attrInt3 -> colStatInt3,
     attrInt4 -> colStatInt4,
@@ -570,6 +579,29 @@ class FilterEstimationSuite extends StatsEstimationTestBase {
       expectedRowCount = 10)
   }
 
+  test("SPARK-59273: CHAR/VARCHAR equality, IN, and range fall back like STRING") {
+    Seq(attrChar -> colStatChar, attrVarchar -> colStatVarchar).foreach {
+      case (attr, colStat) =>
+        validateEstimatedStats(
+          Filter(EqualTo(attr, Literal("A2")), childStatsTestPlan(Seq(attr), 10L)),
+          Seq(attr -> colStat.copy(distinctCount = Some(1), nullCount = Some(0))),
+          expectedRowCount = 1)
+        validateEstimatedStats(
+          Filter(InSet(attr, Set("A0")), childStatsTestPlan(Seq(attr), 10L)),
+          Seq(attr -> colStat.copy(distinctCount = Some(1), nullCount = Some(0))),
+          expectedRowCount = 1)
+        validateEstimatedStats(
+          Filter(LessThan(attr, Literal("A2")), childStatsTestPlan(Seq(attr), 10L)),
+          Seq(attr -> colStat),
+          expectedRowCount = 10)
+    }
+    validateEstimatedStats(
+      Filter(GreaterThan(attrChar, attrVarchar),
+        childStatsTestPlan(Seq(attrChar, attrVarchar), 10L)),
+      Seq(attrChar -> colStatChar, attrVarchar -> colStatVarchar),
+      expectedRowCount = 10)
+  }
+
   test("cint IN (1, 2, 3, 4, 5)") {
     // This is a corner test case.  We want to test if we can handle the case when the number of
     // valid values in IN clause is greater than the number of distinct values for a given column.

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/JoinEstimationSuite.scala

@@ -499,6 +499,10 @@ class JoinEstimationSuite extends StatsEstimationTestBase {
           nullCount = Some(0), avgLen = Some(16), maxLen = Some(16)),
         AttributeReference("cstring", StringType)() -> ColumnStat(distinctCount = Some(1),
           min = None, max = None, nullCount = Some(0), avgLen = Some(3), maxLen = Some(3)),
+        AttributeReference("cchar", CharType(3))() -> ColumnStat(distinctCount = Some(1),
+          min = None, max = None, nullCount = Some(0), avgLen = Some(3), maxLen = Some(3)),
+        AttributeReference("cvarchar", VarcharType(3))() -> ColumnStat(distinctCount = Some(1),
+          min = None, max = None, nullCount = Some(0), avgLen = Some(3), maxLen = Some(3)),
         AttributeReference("cbinary", BinaryType)() -> ColumnStat(distinctCount = Some(1),
           min = None, max = None, nullCount = Some(0), avgLen = Some(3), maxLen = Some(3)),
         AttributeReference("cdate", DateType)() -> ColumnStat(distinctCount = Some(1),
@@ -531,6 +535,10 @@ class JoinEstimationSuite extends StatsEstimationTestBase {
           rowCount = Some(1),
           attributeStats = AttributeMap(Seq(key1 -> columnInfo1(key1), key2 -> columnInfo1(key1))))
         assert(join.stats == expectedStats)
+        if (key1.dataType.isInstanceOf[StringType]) {
+          // UTF8 rows include the average payload length plus base and offset overhead.
+          assert(join.stats.sizeInBytes == 38)
+        }
       }
     }
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/StatsEstimationTestBase.scala

@@ -46,7 +46,7 @@ trait StatsEstimationTestBase extends SparkFunSuite {
 
   def getColSize(attribute: Attribute, colStat: ColumnStat): Long = attribute.dataType match {
     // For UTF8String: base + offset + numBytes
-    case StringType => colStat.avgLen.getOrElse(attribute.dataType.defaultSize.toLong) + 8 + 4
+    case _: StringType => colStat.avgLen.getOrElse(attribute.dataType.defaultSize.toLong) + 8 + 4
     case _ => colStat.avgLen.getOrElse(attribute.dataType.defaultSize)
   }
 

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)