cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4107568122
##########
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:
Confirmed: the dedicated V1 save regression now uses a VARCHAR relation,
binds both scan modes, and checks the retained caches after mutation.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4091094284","thread_id":"inline:4091094284","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -344,7 +344,10 @@ class RelationResolution(
writePrivileges == null && !u.isStreaming
cached <- lookupSharedRelationCache(catalog, ident, t,
tableKey.stateOptions)
} yield {
- val updatedRelation = cached.copy(options = finalOptions)
+ // A shared cache entry may have been analyzed under another
session's CHAR/VARCHAR
+ // policy. Rebind it in this analysis instead of inheriting that
session's scan mode.
+ val updatedRelation =
+ cached.copy(options = finalOptions, charVarcharScanMode = None)
Review Comment:
Confirmed: V2TableReference adaptation now clears the cached relation's
charVarcharScanMode before the current analysis binds it.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4103666672","thread_id":"inline:4103666672","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/LogicalRelation.scala:
##########
@@ -77,6 +80,16 @@ case class LogicalRelation(
case _ => // Do nothing.
}
+ def hasCharVarchar: Boolean = output.exists { attr =>
+ CharVarcharUtils.hasCharVarchar(attr.dataType) ||
+
CharVarcharUtils.getRawType(attr.metadata).exists(CharVarcharUtils.hasCharVarchar)
+ }
+
+ def sameResultWithUnboundCharVarcharScanMode(other: LogicalRelation):
Boolean = {
Review Comment:
Confirmed: the unused LogicalRelation helper is gone, and V1 mutation
invalidation still uses BaseRelation identity.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4103666679","thread_id":"inline:4103666679","verdict_sha256":"1821e60da6a15b8042b8aae53c3feec24e342c5cf8056261204a976d216fdfa6"}
-->
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -348,6 +358,92 @@ 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 cache entries containing the given V2 relation, ignoring scan
mode for the relation
Review Comment:
Confirmed: the default lookup now uses plan.exists across every V2 relation
and still returns at most one descriptor per cache entry.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4103666691","thread_id":"inline:4103666691","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]
