[SPARK-58794][SQL] Empty2Null, text, and Hive prune fallback for CHAR/VARCHAR

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

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"
 }

sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala

@@ -21,6 +21,7 @@ import scala.util.Try
 
 import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException, SparkThrowable}
 import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
+import org.apache.spark.sql.catalyst.analysis.resolver.ResolverGuard
 import org.apache.spark.sql.catalyst.expressions.{Attribute, EqualTo, GreaterThan, Literal, ScalarSubquery, StringRPad}
 import org.apache.spark.sql.catalyst.expressions.Cast.toSQLId
 import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
@@ -1801,6 +1802,16 @@ class BasicCharVarcharTestSuite extends SharedSparkSession {
         SQLConf.ANALYZER_DUAL_RUN_SAMPLE_RATE.key -> "1.0",
         SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_ENABLED_TENTATIVELY.key -> "false",
         SQLConf.ANALYZER_SINGLE_PASS_RESOLVER_EXPOSE_RESOLVER_GUARD_FAILURE.key -> "true") {
+      def assertDualRunEntered(sqlText: String): Unit = {
+        val parsed = spark.sessionState.sqlParser.parsePlan(sqlText)
+        val guard = new ResolverGuard(spark.sessionState.catalogManager)
+        val reason = guard.apply(parsed).planUnsupportedReason
+        assert(reason.isEmpty, s"ResolverGuard skipped dual-run for [$sqlText]: $reason")
+      }
+      assertDualRunEntered("SELECT CAST('ab' AS CHAR(5)) AS c")
+      assertDualRunEntered(
+        "SELECT coalesce(CAST('a' AS CHAR(2)), CAST('bb' AS VARCHAR(4))) AS c")
+
       // CAST / try_cast introduce the type (R3).
       assert(sql("SELECT CAST('ab' AS CHAR(5)) AS c").schema.head.dataType === CharType(5))
       assert(sql("SELECT CAST('hello' AS VARCHAR(5)) AS c").schema.head.dataType ===
@@ -2092,6 +2103,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("<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 {
@@ -2154,6 +2202,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("<ab   >", "<cd>"))
+      }
+      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("<ab  >"))
+        val catalog = spark.read.schema("c VARCHAR(4)").format(format).load(path)
+        assert(catalog.schema.head.dataType === VarcharType(4))
+        checkAnswer(catalog, Row("ab  "))
+      }
     }
   }
 

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") {

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()

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 {