This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 9957fbe39e22 [SPARK-59001][SQL] Empty2Null, text, and Hive prune
fallback for CHAR/VARCHAR
9957fbe39e22 is described below
commit 9957fbe39e22c86f9d0c5f7dba50ce6b7096e2f7
Author: Serge Rielau <[email protected]>
AuthorDate: Thu Aug 27 10:23:31 2026 +0800
[SPARK-59001][SQL] Empty2Null, text, and Hive prune fallback for
CHAR/VARCHAR
### What changes were proposed in this pull request?
[SPARK-59001](https://issues.apache.org/jira/browse/SPARK-59001) (parent
[SPARK-58794](https://issues.apache.org/jira/browse/SPARK-58794)).
Follow-up on first-class CHAR/VARCHAR
(`spark.sql.charVarchar.standardSemantics.enabled`) so a few string-family call
sites treat `CharType` / `VarcharType` like `STRING`.
- `V1Writes` applies `Empty2Null` to nullable CHAR/VARCHAR partition
columns (`dataType.isInstanceOf[StringType]`). `CHAR(n>0)` still pads `''` to
spaces, so the empty CHAR case is `CHAR(0)`.
- The V2 text data source accepts CHAR/VARCHAR as a string-family type.
- Hive metastore filter conversion still refuses CHAR/VARCHAR partition
keys (`varcharKeys` in `SupportedAttribute`: Hive's trailing-blank comparison
is not Spark's). When a predicate mentions such a key,
`prunePartitionsFastFallback` prunes client-side with Spark's own predicates
(CHAR compared without PAD SPACE; the test literal is `'a '` for `CHAR(5)`).
Other empty-filter and MetaException fallbacks still honor
`metastorePartitionPruningFastFallback`.
### Why are the changes needed?
With first-class types, CHAR/VARCHAR stay in the plan instead of being
rewritten to annotated STRING. Equality against `StringType` then skips them:
- empty partition values are not converted to NULL, so they become a
distinct partition directory instead of `__HIVE_DEFAULT_PARTITION__`
- `USING text` rejects a CHAR/VARCHAR schema
- Hive partition filters on CHAR keys fetch every partition
The annotation-skipping idea was tried and dropped: `ApplyCharTypePadding`
uses `__CHAR_VARCHAR_TYPE_STRING` as an idempotence marker, so the annotation
is load-bearing.
### Does this PR introduce _any_ user-facing change?
Yes, only when `spark.sql.charVarchar.standardSemantics.enabled` is true
(still off by default).
- Empty `VARCHAR` / `CHAR(0)` partition values become NULL like STRING.
- `spark.read.schema("value CHAR(n)").text(...)` is accepted.
- Hive CHAR partition filters prune to the matching partitions
(client-side), and `CHAR(5) = 'a'` does not match a stored `'a '` unless the
literal carries the pad or an RTRIM collation is used.
### How was this patch tested?
- `sql/testOnly *CharVarcharTestSuite *V1WriteCommandSuite`: 164 succeeded
- `hive/testOnly *HiveCharVarcharTestSuite`: 58 succeeded
- `hive/testOnly *HivePartitionFilteringSuite*`: 360 succeeded (Hive
2.3-4.1)
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.6
Closes #58255 from srielau/serge-rielau_data/SPARK-58794-p0-followup.
Authored-by: Serge Rielau <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit f22e4b4c0c403001b06c74237974af56d3b99dd5)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../spark/sql/execution/datasources/V1Writes.scala | 2 +-
.../execution/datasources/v2/text/TextTable.scala | 3 +-
.../apache/spark/sql/CharVarcharTestSuite.scala | 41 +++++++++++++
.../datasources/V1WriteCommandSuite.scala | 18 ++++++
.../apache/spark/sql/hive/client/HiveShim.scala | 33 ++++++++--
.../spark/sql/hive/HiveCharVarcharTestSuite.scala | 70 +++++++++++++++++++++-
6 files changed, 159 insertions(+), 8 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/V1Writes.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/V1Writes.scala
index 4493d1a6e689..abd30e9c65a4 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/V1Writes.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/V1Writes.scala
@@ -190,7 +190,7 @@ object V1WritesUtils {
val partitionSet = AttributeSet(partitionColumns)
var needConvert = false
val projectList: Seq[NamedExpression] = output.map {
- case p if partitionSet.contains(p) && p.dataType == StringType &&
p.nullable =>
+ case p if partitionSet.contains(p) &&
p.dataType.isInstanceOf[StringType] && p.nullable =>
needConvert = true
Alias(Empty2Null(p), p.name)()
case attr => attr
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
index d8880b84c621..289c41ca1d0b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/text/TextTable.scala
@@ -46,7 +46,8 @@ case class TextTable(
}
}
- override def supportsDataType(dataType: DataType): Boolean = dataType ==
StringType
+ override def supportsDataType(dataType: DataType): Boolean =
+ dataType.isInstanceOf[StringType]
override def formatName: String = "Text"
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
index 7de934ef2af6..39694ad3d686 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala
@@ -2185,6 +2185,47 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
}
}
}
+
+ test("SPARK-59001: empty CHAR/VARCHAR partition values become null like
STRING") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ // CHAR(n>0) pads '' to spaces; CHAR(0) is the empty CHAR that
empty2null should treat
+ // the same as VARCHAR/STRING.
+ Seq("CHAR(0)", "VARCHAR(5)").foreach { typ =>
+ withTempPath { path =>
+ sql(s"SELECT 0 AS id, CAST('' AS $typ) AS p UNION ALL SELECT 1,
CAST(NULL AS $typ)")
+
.write.mode("overwrite").partitionBy("p").parquet(path.getCanonicalPath)
+ val df = spark.read.parquet(path.getCanonicalPath)
+ checkAnswer(df.where("p IS NULL").select("id"), Seq(Row(0), Row(1)))
+ val dirs = path.listFiles().filterNot(
+ f => f.getName.startsWith(".") || f.getName.startsWith("_"))
+ assert(dirs.length === 1, dirs.map(_.getName).mkString(","))
+ }
+ }
+ }
+ }
+
+ test("SPARK-59001: text datasource accepts CHAR/VARCHAR as a string family
type") {
+ Seq("text", "").foreach { useV1SourceList =>
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true",
+ SQLConf.USE_V1_SOURCE_LIST.key -> useV1SourceList) {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ sql("SELECT CAST('ab' AS CHAR(4)) AS
value").write.mode("overwrite").text(path)
+ val df = spark.read.schema("value CHAR(4)").text(path)
+ assert(df.schema.head.dataType === CharType(4))
+ checkAnswer(df.selectExpr("concat('<', value, '>')"), Row("<ab >"))
+ }
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ sql("SELECT CAST('cd' AS VARCHAR(5)) AS
value").write.mode("overwrite").text(path)
+ val df = spark.read.schema("value VARCHAR(5)").text(path)
+ assert(df.schema.head.dataType === VarcharType(5))
+ checkAnswer(df, Row("cd"))
+ }
+ }
+ }
+ }
}
class FileSourceCharVarcharTestSuite extends CharVarcharTestSuite with
SharedSparkSession {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala
index 9f7e23a5c6de..1f71e77bc08a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/V1WriteCommandSuite.scala
@@ -161,6 +161,24 @@ class V1WriteCommandSuite extends SharedSparkSession with
V1WriteCommandSuiteBas
}
}
+ test("v1 write with CHAR/VARCHAR partition columns applies empty2null") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ // The partition values must vary, otherwise the sort on a foldable key
is pruned and
+ // there is no output ordering left to match.
+ Seq("CHAR(5)", "VARCHAR(5)").foreach { typ =>
+ withPlannedWrite { enabled =>
+ withTable("t") {
+ sql(s"CREATE TABLE t(i INT) USING PARQUET PARTITIONED BY (p $typ)")
+ executeAndCheckOrdering(
+ hasLogicalSort = enabled, orderingMatched = enabled,
hasEmpty2Null = enabled) {
+ sql("INSERT INTO t SELECT i, k FROM t0")
+ }
+ }
+ }
+ }
+ }
+ }
+
test("v1 write with partition, bucketed and sort columns") {
withPlannedWrite { enabled =>
withTable("t") {
diff --git
a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala
b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala
index ef27669f5ba0..bd6ce51966b5 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala
@@ -351,7 +351,12 @@ private[client] class Shim_v2_0 extends Shim with Logging {
val filter = convertFilters(table, predicates)
val partitions =
- if (filter.isEmpty) {
+ if (referencesCharVarcharPartitionKey(catalogTable, predicates)) {
+ // convertFilters skips CHAR/VARCHAR keys. If another conjunct is
supported, its
+ // non-empty filter would otherwise fetch a superset before residual
pruning.
+ prunePartitionsFastFallback(
+ hive, table, catalogTable, predicates, forceClientSide = true)
+ } else if (filter.isEmpty) {
prunePartitionsFastFallback(hive, table, catalogTable, predicates)
} else {
logDebug(s"Hive metastore filter is '$filter'.")
@@ -391,11 +396,29 @@ private[client] class Shim_v2_0 extends Shim with Logging
{
partitions.asScala.toSeq
}
+ private def referencesCharVarcharPartitionKey(
+ catalogTable: CatalogTable,
+ predicates: Seq[Expression]): Boolean = {
+ SQLConf.get.charVarcharStandardSemantics && {
+ val charVarcharPartNames = catalogTable.partitionSchema.fields.collect {
+ case f if CharVarcharUtils.hasCharVarchar(f.dataType) => f.name
+ }
+ charVarcharPartNames.nonEmpty && {
+ val resolver = SQLConf.get.resolver
+ predicates.exists(_.exists {
+ case a: Attribute => charVarcharPartNames.exists(n => resolver(n,
a.name))
+ case _ => false
+ })
+ }
+ }
+ }
+
private def prunePartitionsFastFallback(
hive: Hive,
table: Table,
catalogTable: CatalogTable,
- predicates: Seq[Expression]): java.util.Collection[Partition] = {
+ predicates: Seq[Expression],
+ forceClientSide: Boolean = false): java.util.Collection[Partition] = {
val timeZoneId = SQLConf.get.sessionLocalTimeZone
// Because there is no way to know whether the partition properties has
timeZone,
@@ -408,9 +431,9 @@ private[client] class Shim_v2_0 extends Shim with Logging {
}
}
- if (!SQLConf.get.metastorePartitionPruningFastFallback ||
- predicates.isEmpty ||
- predicates.exists(hasTimeZoneAwareExpression)) {
+ if ((!forceClientSide &&
!SQLConf.get.metastorePartitionPruningFastFallback) ||
+ predicates.isEmpty ||
+ predicates.exists(hasTimeZoneAwareExpression)) {
recordHiveCall()
hive.getAllPartitionsOf(table)
} else {
diff --git
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveCharVarcharTestSuite.scala
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveCharVarcharTestSuite.scala
index 90cb5501ee6f..73a9b312d5ec 100644
---
a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveCharVarcharTestSuite.scala
+++
b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveCharVarcharTestSuite.scala
@@ -17,9 +17,11 @@
package org.apache.spark.sql.hive
-import org.apache.spark.sql.{CharVarcharTestSuite, Row}
+import org.apache.spark.metrics.source.HiveCatalogMetrics
+import org.apache.spark.sql.{CharVarcharTestSuite, QueryTest, Row}
import org.apache.spark.sql.execution.command.CharVarcharDDLTestBase
import org.apache.spark.sql.hive.test.TestHiveSingleton
+import org.apache.spark.sql.internal.SQLConf
class HiveCharVarcharTestSuite extends CharVarcharTestSuite with
TestHiveSingleton {
@@ -91,6 +93,72 @@ class HiveCharVarcharTestSuite extends CharVarcharTestSuite
with TestHiveSinglet
}
}
}
+
+ test("SPARK-59001: CHAR/VARCHAR partition filters prune client-side") {
+ // Keep the relation a HiveTableRelation, otherwise the scan is converted
to a file index
+ // and never reaches HiveShim's metastore filter conversion. Hive
convertFilters still
+ // skips CHAR/VARCHAR keys; pruning is client-side under standardSemantics.
+ withSQLConf(
+ SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true",
+ SQLConf.HIVE_METASTORE_PARTITION_PRUNING.key -> "true",
+ HiveUtils.CONVERT_METASTORE_PARQUET.key -> "false") {
+ val partitionValues = Seq("a", "b", "c", "d", "e")
+
+ def withHivePartTable(partitionType: String)(body: => Unit): Unit = {
+ withTable("std_hive_part") {
+ sql(
+ s"""CREATE TABLE std_hive_part (i INT, p $partitionType)
+ |USING $format PARTITIONED BY (p)""".stripMargin)
+ partitionValues.foreach { v =>
+ sql(s"INSERT INTO std_hive_part PARTITION (p='$v') VALUES (1)")
+ }
+ body
+ }
+ }
+
+ withHivePartTable("CHAR(5)") {
+ HiveCatalogMetrics.reset()
+ // Store assignment pads CHAR(5); compare is not PAD SPACE, so the
literal must match.
+ // checkToRDD = false: checkAnswer would otherwise scan twice and
double the metric.
+ QueryTest.checkAnswer(
+ sql("SELECT i FROM std_hive_part WHERE p = 'a '"),
+ Seq(Row(1)),
+ checkToRDD = false)
+ assert(HiveCatalogMetrics.METRIC_PARTITIONS_FETCHED.getCount === 1)
+ HiveCatalogMetrics.reset()
+ QueryTest.checkAnswer(
+ sql("SELECT i FROM std_hive_part WHERE p = 'a'"),
+ Nil,
+ checkToRDD = false)
+ assert(HiveCatalogMetrics.METRIC_PARTITIONS_FETCHED.getCount === 0)
+ }
+
+ withHivePartTable("VARCHAR(5)") {
+ HiveCatalogMetrics.reset()
+ QueryTest.checkAnswer(
+ sql("SELECT i FROM std_hive_part WHERE p = 'a'"),
+ Seq(Row(1)),
+ checkToRDD = false)
+ assert(HiveCatalogMetrics.METRIC_PARTITIONS_FETCHED.getCount === 1)
+ }
+
+ // A supported conjunct must not bypass client-side pruning of the CHAR
predicate.
+ withTable("std_hive_part") {
+ sql(
+ s"""CREATE TABLE std_hive_part (i INT, ds INT, p CHAR(5))
+ |USING $format PARTITIONED BY (ds, p)""".stripMargin)
+ partitionValues.foreach { v =>
+ sql(s"INSERT INTO std_hive_part PARTITION (ds=1, p='$v') VALUES (1)")
+ }
+ HiveCatalogMetrics.reset()
+ QueryTest.checkAnswer(
+ sql("SELECT i FROM std_hive_part WHERE ds = 1 AND p = 'a '"),
+ Seq(Row(1)),
+ checkToRDD = false)
+ assert(HiveCatalogMetrics.METRIC_PARTITIONS_FETCHED.getCount === 1)
+ }
+ }
+ }
}
class HiveCharVarcharDDLTestSuite extends CharVarcharDDLTestBase with
TestHiveSingleton {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
