cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4009041874
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +349,47 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
recacheByCondition(spark, _.plan.exists(_.sameResult(normalized)))
}
+ /**
+ * Re-caches every entry whose plan contains a [[LogicalRelation]] for
`relation`.
+ * Unlike [[recacheByPlan]], this ignores CHAR/VARCHAR scan-mode identity so
a V1 write
+ * invalidates preserve-only, standard, and unbound cache entries for that
BaseRelation.
+ */
+ def recacheByV1Relation(spark: SparkSession, relation: BaseRelation): Unit =
{
+ recacheByCondition(spark, cd => cd.plan.exists {
+ case logical: LogicalRelation => logical.relation == relation
+ case _ => false
+ })
+ }
+
+ /**
+ * Re-caches every entry whose plan contains the given catalog-less
[[DataSourceV2Relation]].
+ * The scan mode is ignored only for this mutation-specific match so an
unbound write target
+ * invalidates preserve-native and standard cache entries without weakening
normal cache identity.
+ */
+ def recacheByV2Relation(spark: SparkSession, relation:
DataSourceV2Relation): Unit = {
+ val unboundRelation = relation.copy(charVarcharScanMode = None)
+ recacheByCondition(spark, cd => cd.plan.exists {
+ case cached: DataSourceV2Relation =>
+ cached.copy(charVarcharScanMode = None).sameResult(unboundRelation)
+ case _ => false
+ })
+ }
+
+ /**
+ * Looks up direct cache entries for a V2 table mutation while ignoring only
their analyzed
+ * CHAR/VARCHAR scan mode. Normal cache substitution remains mode-sensitive.
+ */
+ def lookupCachedDataByV2Relation(relation: DataSourceV2Relation):
Seq[CachedData] = {
+ val unboundRelation = relation.copy(charVarcharScanMode = None)
+ cachedData.filter { cd =>
+ EliminateSubqueryAliases(cd.plan) match {
Review Comment:
**Non-blocking (P2):** This still misses direct caches of tables that
actually contain CHAR/VARCHAR columns. ApplyCharTypePadding puts a Project
above the cleaned DataSourceV2Relation, and EliminateSubqueryAliases leaves
that Project in place, so this match falls through; the later name-based
invalidation removes the old cache and rename restores no variant or
StorageLevel. The new multi-mode rename test remains green because its
INT/STRING schema never creates that Project. Could the direct-cache matcher
recognize only the analyzer-owned CHAR/VARCHAR wrapper, while retaining the
dependent/time-travel exclusions, and exercise it with a real CHAR or VARCHAR
column?
**Recommended change:** Teach the direct V2 cache matcher to unwrap only the
analyzer-owned CHAR/VARCHAR projection shape around the target relation,
preserve mode-insensitive relation matching, and add V2 cache-rename coverage
using real CHAR/VARCHAR columns for every restored scan mode and storage level.
**Why this works:** Recognize the exact ApplyCharTypePadding projection over
a single DataSourceV2Relation as a direct cache key, compare its underlying
relation after clearing charVarcharScanMode, and continue rejecting arbitrary
projections, dependent queries, and time-travel relations. Snapshot and restore
the descriptors already associated with every matched cache.
**Scope:** sql/core/src/main/scala/org/apache/spark/sql/execution,
sql/core/src/test/scala/org/apache/spark/sql
**Compatibility:** Bare-relation direct caches continue to survive V2
rename, while ordinary cache lookup remains mode-sensitive and rename does not
restore dependent or time-travel entries.
**Risks:** An overly broad Project match can promote dependent query caches
to table caches after rename. An overly shape-specific match can miss
legitimate preserve-native or standard-semantics variants as analyzer
expressions evolve.
**Constraints:** Normal cache substitution must remain scan-mode-sensitive.
Dependent query and time-travel cache entries must remain excluded from rename
restoration. Each restored cache must retain its original StorageLevel and
CHAR/VARCHAR scan mode.
**Success:** Direct V2 table caches whose key is wrapped by CHAR/VARCHAR
padding are discovered before rename invalidation. Every preserve-native and
standard-semantics variant is cached under the destination identifier with its
original storage level. Dependent query and time-travel caches are not restored
as direct caches of the destination table.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RenameTableExec.scala:
##########
@@ -49,9 +50,11 @@ case class RenameTableExec(
} else newIdent
catalog.renameTable(oldIdent, qualifiedNewIdent)
Review Comment:
**Non-blocking (P2):** This reloads the same destination table once per
cache variant after renameTable has already committed. With PreserveNative and
SparkStandard caches, that adds avoidable remote catalog latency; if a later
load fails, the command reports failure with the table already renamed and only
a prefix of the caches restored. Could we load qualifiedNewIdent once when
oldCaches is non-empty and reuse that Table for every mode/StorageLevel
descriptor?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala:
##########
@@ -437,6 +437,9 @@ object OrcUtils extends Logging {
s"array<${getOrcSchemaString(a.elementType)}>"
case m: MapType =>
s"map<${getOrcSchemaString(m.keyType)},${getOrcSchemaString(m.valueType)}>"
+ // Keep Spark responsible for CHAR/VARCHAR assignment and scan checks.
Native ORC
+ // CHAR/VARCHAR would truncate or pad before Spark can validate the
original value.
+ case _: CharType | _: VarcharType => StringType.catalogString
Review Comment:
Confirmed: the row decoder now accepts all StringType subtypes recursively
and the V1/V2 row and vector coverage exercises the requested paths. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3882881464","thread_id":"inline:3882881464","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/test/scala/org/apache/spark/sql/CharVarcharTestSuite.scala:
##########
@@ -1697,33 +1697,94 @@ class BasicCharVarcharTestSuite extends
SharedSparkSession {
sql("DROP TEMPORARY FUNCTION IF EXISTS std_char_param")
sql("DROP TEMPORARY FUNCTION IF EXISTS std_varchar_param")
}
+ }
+ }
- // ORC catalog tables stamp the catalyst type so typeof survives
write/read.
- withTable("std_orc") {
- sql("CREATE TABLE std_orc (c CHAR(5), v VARCHAR(5)) USING orc")
- sql("INSERT INTO std_orc VALUES ('ab', 'cd')")
- assert(spark.table("std_orc").schema.map(_.dataType) ===
- Seq(CharType(5), VarcharType(5)))
- checkAnswer(
- sql("SELECT concat('<', c, '>'), concat('<', v, '>') FROM std_orc"),
- Row("<ab >", "<cd>"))
+ test("SPARK-58814: major formats preserve CHAR/VARCHAR schemas and values") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ Seq("parquet", "orc").foreach { format =>
+ Seq("v1" -> format, "v2" -> "").foreach { case (sourceVersion,
useV1List) =>
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> useV1List) {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ val input = spark.range(1).selectExpr(
+ "cast('ab' AS CHAR(4)) AS c",
+ "cast('xy' AS VARCHAR(3)) AS v",
+ "named_struct('c', cast('z' AS CHAR(2))) AS s",
+ "array(cast('q' AS VARCHAR(2))) AS a",
+ "map(cast('k' AS CHAR(2)), cast('v' AS VARCHAR(2))) AS m")
+ input.write.mode("overwrite").format(format).save(path)
+
+ val readBack = spark.read.format(format).load(path)
+ assert(DataType.equalsIgnoreNullability(readBack.schema,
input.schema),
+ s"$format $sourceVersion lost CHAR/VARCHAR schema")
+ checkAnswer(
+ readBack.selectExpr("concat('<', c, '>')", "v", "concat('<',
s.c, '>')"),
Review Comment:
Confirmed: the ORC assertions now materialize the collection fields and
cover oversized struct, array, map-key, and map-value positions. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3882881471","thread_id":"inline:3882881471","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala:
##########
@@ -437,6 +437,11 @@ object OrcUtils extends Logging {
s"array<${getOrcSchemaString(a.elementType)}>"
case m: MapType =>
s"map<${getOrcSchemaString(m.keyType)},${getOrcSchemaString(m.valueType)}>"
+ // Under standard semantics, keep Spark responsible for CHAR/VARCHAR
assignment and scan
+ // checks. Native ORC would truncate or pad before Spark can validate the
original value.
+ // Preserve-only mode retains the native constrained schema and its legacy
enforcement.
+ case _: CharType | _: VarcharType if
SQLConf.get.charVarcharStandardSemantics =>
Review Comment:
Confirmed: persisted views now retain the analysis-bound scan mode across
caller configuration changes. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3892187814","thread_id":"inline:3892187814","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -3724,6 +3724,82 @@ abstract class AvroSuite
}
}
+ test("SPARK-58814: Avro infers nested CHAR/VARCHAR schema and values") {
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key -> "true") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ val input = spark.range(1).selectExpr(
+ "cast('ab' AS CHAR(4)) AS c",
+ "cast('xy' AS VARCHAR(3)) AS v",
+ "named_struct('c', cast('z' AS CHAR(2))) AS s",
+ "array(cast('q' AS VARCHAR(2))) AS a",
+ "map(cast('k' AS CHAR(2)), cast('v' AS VARCHAR(2))) AS m")
+ input.write.mode("overwrite").format("avro").save(path)
+
+ val readBack = spark.read.format("avro").load(path)
+ assert(DataType.equalsIgnoreNullability(readBack.schema, input.schema))
+ checkAnswer(
+ readBack.selectExpr("concat('<', c, '>')", "v", "concat('<', s.c,
'>')"),
Review Comment:
Confirmed: the Avro round trip now materializes and asserts both collection
fields. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3892187822","thread_id":"inline:3892187822","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala:
##########
@@ -427,16 +427,29 @@ object OrcUtils extends Logging {
* Given a `StructType` object, this methods converts it to corresponding
string representation
* in ORC.
*/
- def getOrcSchemaString(dt: DataType): String = dt match {
+ def getOrcSchemaString(dt: DataType): String = {
+ getOrcSchemaString(dt, SQLConf.get.charVarcharStandardSemantics)
+ }
+
+ private def getOrcSchemaString(
+ dt: DataType,
+ charVarcharStandardSemantics: Boolean): String = dt match {
case s: StructType =>
val fieldTypes = s.fields.map { f =>
- s"${quoteIdentifier(f.name)}:${getOrcSchemaString(f.dataType)}"
+ s"${quoteIdentifier(f.name)}:" +
+ s"${getOrcSchemaString(f.dataType, charVarcharStandardSemantics)}"
}
s"struct<${fieldTypes.mkString(",")}>"
case a: ArrayType =>
- s"array<${getOrcSchemaString(a.elementType)}>"
+ s"array<${getOrcSchemaString(a.elementType,
charVarcharStandardSemantics)}>"
case m: MapType =>
-
s"map<${getOrcSchemaString(m.keyType)},${getOrcSchemaString(m.valueType)}>"
+ s"map<${getOrcSchemaString(m.keyType, charVarcharStandardSemantics)}," +
+ s"${getOrcSchemaString(m.valueType, charVarcharStandardSemantics)}>"
+ // Under standard semantics, keep Spark responsible for CHAR/VARCHAR
assignment and scan
+ // checks. Native ORC would truncate or pad before Spark can validate the
original value.
+ // Preserve-only mode retains the native constrained schema and its legacy
enforcement.
+ case _: CharType | _: VarcharType if charVarcharStandardSemantics =>
Review Comment:
Confirmed: the preserve-native ORC round trip writes first-class constrained
types and verifies inferred types plus native padding and truncation in V1 and
V2. Thanks.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:3898132494","thread_id":"inline:3898132494","verdict_sha256":"9e8299073fe3a3fd317d35d14ddb819ddbbaf0c580a01f233a538f0836de3d9e"}
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
