srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4097792411
##########
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:
Addressed in d161b01b76c.
Direct V2 rename lookup no longer matches CACHE TABLE display names. It
walks the cached plan and keeps a cache only when it is a bare
`DataSourceV2Relation` or an analyzer-generated CHAR/VARCHAR padding `Project`
over that relation. The V2 rename test now caches with the short identifier
`tbl` after switching the current catalog, so this path is covered.
A broader CacheManager descriptor (resolved origin plus captured
plan/mode/storage level for V1/V2/Hive) remains in
[SPARK-59751](https://issues.apache.org/jira/browse/SPARK-59751).
##########
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:
Addressed in d161b01b76c.
Rename restoration no longer uses `withSQLConf` on the caller session. It
clones the session, sets the captured CHAR/VARCHAR mode on the clone, and
recaches through that clone. SharedState/CacheManager stay shared, so the
caller SQLConf is unchanged.
Isolated restoration as a general CacheManager contract (not just this V2
rename path) remains in
[SPARK-59751](https://issues.apache.org/jira/browse/SPARK-59751).
##########
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
Review Comment:
Addressed in d161b01b76c.
The helper Scaladoc now describes both call forms: default lookup returns
any cache containing the relation, and `directNamedCacheOnly = true` returns
only direct table caches, including analyzer-generated CHAR/VARCHAR projections.
##########
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:
Addressed in d161b01b76c.
The Hive ORC cache test now loops `CONVERT_METASTORE_ORC` true and false, so
it covers converted `LogicalRelation` and native `HiveTableRelation`. Converted
standard-semantics still expects `EXCEED_LIMIT_LENGTH` for `"abcdef"` into
`CHAR(4)`; native Hive truncates to `"abcd"`. Both paths assert the two bound
caches stay distinct.
A unified V1/V2/Hive mutation descriptor remains in
[SPARK-59751](https://issues.apache.org/jira/browse/SPARK-59751).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SaveIntoDataSourceCommandSuite.scala:
##########
@@ -69,7 +70,29 @@ class SaveIntoDataSourceCommandSuite extends
SharedSparkSession {
saveIntoDataSource(2)
checkAnswer(loadData, Row(0) :: Row(1) :: Nil)
+ spark.catalog.clearCache()
FakeV1DataSource.data = null
+
+ // The two bound scan modes are part of LogicalRelation identity, so
recache must match the
Review Comment:
Addressed in d161b01b76c.
The original SPARK-39952 LONG save test is unchanged. A separate test now
uses `FakeRelation` with `VARCHAR(4)`, asserts `PreserveNative` and
`SparkStandard` are bound, then overwrites once and checks both caches refresh.
Further CacheManager lifecycle unification is in
[SPARK-59751](https://issues.apache.org/jira/browse/SPARK-59751).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoDataSourceCommand.scala:
##########
@@ -45,7 +45,8 @@ case class InsertIntoDataSourceCommand(
// Re-cache all cached plans(including this relation itself, if it's
cached) that refer to this
// data source relation.
- sparkSession.sharedState.cacheManager.recacheByPlan(sparkSession,
logicalRelation)
+ sparkSession.sharedState.cacheManager.recacheByV1Relation(
Review Comment:
Addressed in d161b01b76c.
The insert recache test now uses a CHAR/VARCHAR `InsertableRelation`
fixture, binds both `PreserveNative` and `SparkStandard`, performs one
overwrite, and asserts both caches remain and return the replacement rows. The
old unbound `jsonTable` path is no longer the coverage for this change.
--
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]
