You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by dt...@apache.org on 2026/09/08 19:12:38 UTC
(spark) branch master updated: [SPARK-59273][SQL] Complete CHAR/VARCHAR support at core execution boundaries
This is an automated email from the ASF dual-hosted git repository.
dtenedor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 51f54b07a3ff [SPARK-59273][SQL] Complete CHAR/VARCHAR support at core execution boundaries
51f54b07a3ff is described below
commit 51f54b07a3fffb8156257e1b2c4037ee7fe49987
Author: Serge Rielau <se...@rielau.com>
AuthorDate: Tue Sep 8 12:12:26 2026 -0700
[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 <se...@rielau.com>
Signed-off-by: Daniel Tenedorio <da...@databricks.com>
---
.../logical/statsEstimation/EstimationUtils.scala | 2 +-
.../logical/statsEstimation/FilterEstimation.scala | 10 ++--
.../logical/statsEstimation/ValueInterval.scala | 2 +-
.../statsEstimation/FilterEstimationSuite.scala | 32 +++++++++++
.../statsEstimation/JoinEstimationSuite.scala | 8 +++
.../statsEstimation/StatsEstimationTestBase.scala | 2 +-
.../spark/sql/classic/DataFrameNaFunctions.scala | 2 +-
.../org/apache/spark/sql/execution/Columnar.scala | 2 +-
.../execution/command/AnalyzeColumnCommand.scala | 1 -
.../execution/datasources/PartitioningUtils.scala | 2 +-
.../datasources/jdbc/JDBCValueGetter.scala | 2 +-
.../sql/execution/datasources/jdbc/JdbcUtils.scala | 12 ++--
.../spark/sql/DataFrameNaFunctionsSuite.scala | 13 ++++-
.../spark/sql/StatisticsCollectionSuite.scala | 44 ++++++++++++++
.../sql/execution/RowToColumnConverterSuite.scala | 16 ++++++
.../parquet/ParquetPartitionDiscoverySuite.scala | 28 +++++++++
.../org/apache/spark/sql/jdbc/JDBCSuite.scala | 67 ++++++++++++++++++++++
17 files changed, 226 insertions(+), 19 deletions(-)
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 7083014f1f38..c107e5428852 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 814c74d16015..125ac22cbc3d 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 4d8dc94f501a..8394ffdbb34b 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 6bd8f7d24ca2..64e91b8cacdd 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 cdfc863cc021..68567541e197 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 17d182ba172b..2564effbf25a 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 b4da4aae07c4..4bd4d8be8157 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 c6323575c9a8..7055fcc631e5 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 924e8b820a7c..1bf25c129e61 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 53dac003754c..c585716b1d53 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 64debbd43a31..dee5867c1e69 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 b513490b8c0b..39c606b0ce1d 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 1282eec920bf..43e2673bdd4a 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 5222d5ce2665..45b1f7931ea1 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 d240a5ab09f8..dfcbc01b5f89 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 2164f4226aa2..aa32e4a3d1ec 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 b1114194f9c7..c79fe8dee0a8 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.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org