diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala index 7083014f1f38f..c107e54288529 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/EstimationUtils.scala +++ b/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 _ => diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala index 814c74d160156..125ac22cbc3d8 100755 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/FilterEstimation.scala +++ b/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) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/ValueInterval.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/ValueInterval.scala index 4d8dc94f501a7..8394ffdbb34b4 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/ValueInterval.scala +++ b/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( diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/FilterEstimationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/FilterEstimationSuite.scala index 6bd8f7d24ca2a..64e91b8cacddd 100755 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/FilterEstimationSuite.scala +++ b/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. diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/JoinEstimationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/JoinEstimationSuite.scala index cdfc863cc0212..68567541e1970 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/JoinEstimationSuite.scala +++ b/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) + } } } } diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/StatsEstimationTestBase.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/StatsEstimationTestBase.scala index 17d182ba172ba..2564effbf25a4 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/statsEstimation/StatsEstimationTestBase.scala +++ b/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) } 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..45b1f7931ea17 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,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) { 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.