cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4107565929
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala:
##########
@@ -115,7 +115,10 @@ case class DataSourceV2Relation(
catalog: Option[CatalogPlugin],
identifier: Option[Identifier],
options: CaseInsensitiveStringMap,
- timeTravelSpec: Option[TimeTravelSpec] = None)
+ timeTravelSpec: Option[TimeTravelSpec] = None,
+ // Bound at analysis so sameResult / cache reuse distinguish preserve-only
vs standard
+ // CHAR/VARCHAR scans. None means the relation was not analyzed under
first-class types.
+ charVarcharScanMode: Option[CharVarcharScanMode] = None)
Review Comment:
The catalog-less callback branch is now present, but the regression still
reads newly constructed DataFrames rather than executing preserveRead and
standardRead after commit; those retained cached DataFrames need to be the
failure signal.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4015284304","thread_id":"inline:4015284304","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcPartitionReaderFactory.scala:
##########
@@ -49,6 +49,7 @@ import org.apache.spark.util.ArrayImplicits._
* @param readDataSchema Required data schema in the batch scan.
* @param partitionSchema Schema of partitions.
* @param options Options for parsing ORC files.
+ * @param charVarcharStandardSemantics CHAR/VARCHAR semantics bound during
analysis.
Review Comment:
Confirmed: the parameter documentation now defines SparkStandard,
PreserveNative, and the unbound native-decoding fallback explicitly.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4015284308","thread_id":"inline:4015284308","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala:
##########
@@ -41,7 +41,10 @@ case class LogicalRelation(
output: Seq[AttributeReference],
catalogTable: Option[CatalogTable],
override val isStreaming: Boolean,
- @transient stream: Option[SparkDataStream])
+ @transient stream: Option[SparkDataStream],
+ // Bound at analysis so sameResult / cache reuse distinguish preserve-only
vs standard
+ // CHAR/VARCHAR scans. None means the relation was not analyzed under
first-class types.
+ charVarcharScanMode: Option[CharVarcharScanMode])
Review Comment:
The Hive refresh dispatch is present, but the new test calls clearCache
immediately before REFRESH TABLE. Please keep the pre-refresh caches
materialized through refresh and assert both variants observe the rewritten
data.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4015284315","thread_id":"inline:4015284315","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +358,81 @@ 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 = {
+ recacheByCondition(spark, cd => cd.plan.exists {
+ case cached: DataSourceV2Relation =>
+ cached.sameResultWithUnboundCharVarcharScanMode(relation)
+ case _ => false
+ })
+ }
+
+ /**
+ * Removes cache entries whose plans contain the given catalog-less
+ * [[DataSourceV2Relation]]. The scan mode is ignored only for this
mutation-specific match.
+ */
+ def uncacheByV2Relation(
+ spark: SparkSession,
+ relation: DataSourceV2Relation,
+ cascade: Boolean,
+ blocking: Boolean = false): Unit = {
+ uncacheByCondition(
+ spark,
+ {
+ case cached: DataSourceV2Relation =>
+ cached.sameResultWithUnboundCharVarcharScanMode(relation)
+ case _ => false
+ },
+ cascade,
+ blocking)
+ }
+
+ /**
+ * Describes direct named cache entries for a V2 table. Matching the cache's
table name excludes
+ * dependent query caches while retaining the padding project of a
standard-semantics scan.
+ * Scan mode is ignored for the relation match.
+ */
+ def lookupCacheDescriptorsByV2Relation(
+ relation: DataSourceV2Relation,
+ directNamedCacheOnly: Boolean = false): Seq[TableCacheDescriptor] = {
+ val directCacheName = for {
+ catalog <- relation.catalog
+ ident <- relation.identifier
+ } yield ident.toQualifiedNameParts(catalog).quoted
+ cachedData.flatMap { cd =>
+ val isDirectCache = !directNamedCacheOnly ||
+ cd.cachedRepresentation.cacheBuilder.tableName == directCacheName
Review Comment:
Confirmed: direct V2 rename lookup is now structural and no longer compares
the cache's short display label with the qualified catalog name.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4091094244","thread_id":"inline:4091094244","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/RenameTableExec.scala:
##########
@@ -49,14 +53,43 @@ case class RenameTableExec(
} else newIdent
catalog.renameTable(oldIdent, qualifiedNewIdent)
- optOldStorageLevel.foreach { oldStorageLevel =>
- val tbl = catalog.loadTable(qualifiedNewIdent)
- val newRelation = DataSourceV2Relation.create(tbl, Some(catalog),
Some(qualifiedNewIdent))
- cacheTable(
- session,
- newRelation,
- Some(qualifiedNewIdent.quoted), oldStorageLevel)
+ val table = if (oldCaches.nonEmpty) {
+ Some(catalog.loadTable(qualifiedNewIdent))
+ } else {
+ None
+ }
+ oldCaches.foreach { cache =>
+ val rewritten = cache.plan.transformUp {
+ case relation: DataSourceV2Relation
+ if relation.catalog.contains(catalog) &&
relation.identifier.contains(oldIdent) =>
+ val restored = relation.copy(
+ table = table.get,
+ catalog = Some(catalog),
+ identifier = Some(qualifiedNewIdent))
+ restored.copyTagsFrom(relation)
+ restored.setAnalyzed()
+ restored
+ }
+ withCharVarcharScanModeConf(cache.charVarcharScanMode) {
+ cacheTable(
+ session,
+ rewritten,
+ Some(qualifiedNewIdent.quoted), cache.storageLevel)
+ }
}
Seq.empty
}
+
+ // Re-cache under the mode that produced the original plan so CHAR/VARCHAR
output is legal.
+ private def withCharVarcharScanModeConf[T](mode:
Option[CharVarcharScanMode])(body: => T): T = {
+ mode match {
+ case Some(CharVarcharScanMode.SparkStandard) =>
+ withSQLConf(SQLConf.CHAR_VARCHAR_STANDARD_SEMANTICS.key ->
"true")(body)
Review Comment:
Confirmed: rename restoration now configures a cloned session per descriptor
and does not temporarily mutate the caller session's CHAR/VARCHAR setting.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4091094256","thread_id":"inline:4091094256","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala:
##########
@@ -88,6 +91,60 @@ class CachedTableSuite extends QueryTest with
TestHiveSingleton {
sql("DROP TABLE cachedTable")
}
+ test("SPARK-58814: Hive ORC caches do not cross CHAR/VARCHAR scan modes") {
+ val tableName = "hive_char_cache"
+ withTempPath { path =>
+ Seq("abcdef").toDF("v").write.orc(path.getCanonicalPath)
+ withTable(tableName) {
+ sql(
+ s"""CREATE EXTERNAL TABLE $tableName (v CHAR(4))
+ |STORED AS ORC LOCATION '${path.toURI}'""".stripMargin)
+
+ withSQLConf(
+ HiveUtils.CONVERT_METASTORE_ORC.key -> "true",
Review Comment:
Confirmed: the Hive cache test now runs with CONVERT_METASTORE_ORC both true
and false, so it reaches converted LogicalRelation and native HiveTableRelation
paths.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4091094275","thread_id":"inline:4091094275","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
--
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]
