From bdcbdae8559ed79b530aa5b1c9d2aba5da6c1d3e Mon Sep 17 00:00:00 2001 From: srielau Date: Mon, 24 Aug 2026 19:34:01 +0000 Subject: [PATCH] [SPARK-58794][SQL] Empty2Null, text, and Hive prune fallback for CHAR/VARCHAR Treat CHAR/VARCHAR as a string family for partition empty-to-null and the text data source. Hive metastore filter conversion still excludes these keys, so under standard semantics prune them client-side instead. --- .../sql/execution/datasources/V1Writes.scala | 2 +- .../datasources/v2/text/TextTable.scala | 3 +- .../spark/sql/CharVarcharTestSuite.scala | 57 +++++++++++++++++++ .../datasources/V1WriteCommandSuite.scala | 18 ++++++ .../spark/sql/hive/client/HiveShim.scala | 9 ++- .../sql/hive/HiveCharVarcharTestSuite.scala | 37 ++++++++++++ 6 files changed, 123 insertions(+), 3 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 4493d1a6e6895..abd30e9c65a4d 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 d8880b84c6211..289c41ca1d0b2 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 a9e3c0626e0fd..19c1417c1ea23 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 @@ -2157,6 +2157,43 @@ class BasicCharVarcharTestSuite extends SharedSparkSession { } } } + + test("SPARK-58794: 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-58794: text datasource accepts CHAR/VARCHAR as a string family type") { + withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") { + 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("")) + } + 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 { @@ -2219,6 +2256,26 @@ class FileSourceCharVarcharTestSuite extends CharVarcharTestSuite with SharedSpa checkAnswer(sql("SELECT * FROM t"), Row("12")) } } + // Catalog write/read and file inference both keep CHAR/VARCHAR. + withTable("std_parquet") { + sql(s"CREATE TABLE std_parquet (c CHAR(5), v VARCHAR(5)) USING $format") + sql("INSERT INTO std_parquet VALUES ('ab', 'cd')") + assert(spark.table("std_parquet").schema.map(_.dataType) === + Seq(CharType(5), VarcharType(5))) + checkAnswer( + sql("SELECT concat('<', c, '>'), concat('<', v, '>') FROM std_parquet"), + Row("", "")) + } + withTempPath { dir => + val path = dir.getCanonicalPath + sql("SELECT CAST('ab' AS CHAR(4)) AS c").write.mode("overwrite").format(format).save(path) + val inferred = spark.read.format(format).load(path) + assert(inferred.schema.head.dataType === CharType(4)) + checkAnswer(inferred.selectExpr("concat('<', c, '>')"), Row("")) + val catalog = spark.read.schema("c VARCHAR(4)").format(format).load(path) + assert(catalog.schema.head.dataType === VarcharType(4)) + checkAnswer(catalog, Row("ab ")) + } } } 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 9f7e23a5c6de8..1f71e77bc08a2 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 32d8928836976..b3fd4094979f6 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 @@ -408,7 +408,14 @@ private[client] class Shim_v2_0 extends Shim with Logging { } } - if (!SQLConf.get.metastorePartitionPruningFastFallback || + // CHAR/VARCHAR partition keys are excluded from the metastore filter (see + // SupportedAttribute), because Hive compares them with its own trailing-blank rules. Under + // standard semantics that would leave such a query fetching every partition, so prune on the + // client instead, where Spark's own comparison semantics apply. + val charVarcharPartitionKey = SQLConf.get.charVarcharStandardSemantics && + catalogTable.partitionSchema.exists(f => CharVarcharUtils.hasCharVarchar(f.dataType)) + + if ((!SQLConf.get.metastorePartitionPruningFastFallback && !charVarcharPartitionKey) || predicates.isEmpty || predicates.exists(hasTimeZoneAwareExpression)) { recordHiveCall() 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 90cb5501ee6f6..2a98954ae7b5b 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.metrics.source.HiveCatalogMetrics import org.apache.spark.sql.{CharVarcharTestSuite, 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,41 @@ class HiveCharVarcharTestSuite extends CharVarcharTestSuite with TestHiveSinglet } } } + + test("SPARK-58794: CHAR partition filters use the same Hive prune path as STRING") { + // Keep the relation a HiveTableRelation, otherwise the scan is converted to a file index + // and never reaches HiveShim's metastore filter conversion. + 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 partitionsFetched(partitionType: String, literal: String): Long = { + var fetched = 0L + 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)") + } + HiveCatalogMetrics.reset() + checkAnswer(sql(s"SELECT i FROM std_hive_part WHERE p = $literal"), Row(1)) + fetched = HiveCatalogMetrics.METRIC_PARTITIONS_FETCHED.getCount + } + fetched + } + + val stringFetched = partitionsFetched("STRING", "'a'") + // Standard semantics compare CHAR without PAD SPACE, so the literal carries the pad. + val charFetched = partitionsFetched("CHAR(5)", "'a '") + assert(stringFetched < partitionValues.length, + s"STRING baseline did not prune: fetched $stringFetched of ${partitionValues.length}") + assert(charFetched === stringFetched, + s"CHAR fetched $charFetched partitions but STRING fetched $stringFetched") + } + } } class HiveCharVarcharDDLTestSuite extends CharVarcharDDLTestBase with TestHiveSingleton {