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

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)

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,50 @@ 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("SPARK-59273: CBO plans CHAR/VARCHAR predicates after ANALYZE") {
+    withSQLConf(
+        SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true",
+        SQLConf.CBO_ENABLED.key -> "true") {
+      val tableName = "char_varchar_cbo_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")
+
+        // CBO FilterEstimation used to throw a MatchError on CharType/VarcharType after ANALYZE.
+        sql(s"SELECT c FROM $tableName WHERE c = 'a'").collect()
+        sql(s"SELECT c FROM $tableName WHERE c IN ('a  ', 'bb ')").collect()
+        sql(s"SELECT v FROM $tableName WHERE v IN ('x', 'yz')").collect()
+        sql(s"SELECT v FROM $tableName WHERE v > 'x'").collect()
+
+        checkAnswer(
+          sql(s"SELECT v FROM $tableName WHERE v IN ('x', 'yz')"),
+          Row("x") :: Row("yz") :: Nil)
+        checkAnswer(
+          sql(s"SELECT v FROM $tableName WHERE v > 'x'"),
+          Row("yz"))
+      }
+    }
+  }
+
   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)](