diff --git a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameNaFunctions.scala b/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameNaFunctions.scala index b4da4aae07c4a..4bd4d8be81570 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/classic/DataFrameNaFunctions.scala +++ b/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") diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala index c6323575c9a86..7055fcc631e52 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala +++ b/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 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala index 924e8b820a7c3..1bf25c129e61b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeColumnCommand.scala +++ b/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 } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala index 53dac003754cd..c585716b1d53f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala +++ b/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) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala index 64debbd43a31a..dee5867c1e69a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCValueGetter.scala +++ b/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] { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala index b513490b8c0b9..39c606b0ce1d8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala +++ b/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)) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameNaFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameNaFunctionsSuite.scala index 1282eec920bf2..43e2673bdd4a7 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameNaFunctionsSuite.scala +++ b/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, diff --git a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala index 5222d5ce26658..a1a0be96fcbe1 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/StatisticsCollectionSuite.scala +++ b/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) { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/RowToColumnConverterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/RowToColumnConverterSuite.scala index d1d8e89de3912..7b7d45c6de426 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/RowToColumnConverterSuite.scala +++ b/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))) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala index 2164f4226aa24..aa32e4a3d1ec2 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala +++ b/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)]( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala index b1114194f9c74..c79fe8dee0a8c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala +++ b/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.