cloud-fan commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4091094244
##########
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:
**Non-blocking (P2):** A direct cache created as CACHE TABLE t keeps the
caller's short label, while this comparison uses the fully qualified catalog
name. The lookup returns no descriptor, then rename invalidates the old entry
and never restores it at the new name or storage level.
**Recommended change:** Capture resolved direct V2 table origin when a named
table cache is created, then match rename descriptors by that origin while
preserving captured plan, mode, and storage level.
**Why this works:** A direct-origin descriptor uses the resolved catalog and
identifier as the rename key, so all spellings of the same table select the
same direct cache entries; dependent query entries lack that direct origin.
**Scope:** Make V2 direct-cache identity stable across name spellings and
rename restoration.
**Compatibility:** Ordinary cache substitution remains scan-mode-sensitive;
mutation lookup ignores only mode when identifying the same table.
**Risks:** Changing descriptor creation must preserve cache keys for
ordinary mode-sensitive substitution.
**Constraints:** Exclude dependent query and time-travel cache keys from
direct rename restoration.
**Success:** A direct V2 cache made through unqualified SQL or
Catalog.cacheTable remains cached after rename at its prior storage level.
Dependent query caches are invalidated without being recreated as direct table
caches.
##########
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:
**Nit (P3):** The direct-only description applies only when
directNamedCacheOnly is true. With the default false, a dependent cached query
containing this relation is returned too, so the comment should describe both
call forms.
##########
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:
**Non-blocking (P2):** This restoration temporarily changes the shared
session's CHAR/VARCHAR setting. Another operation on the same SparkSession can
analyze under that transient mode, or its config update can be lost when
withSQLConf restores the prior value.
**Recommended change:** Replay each captured cache plan and mode using an
isolated session configuration that shares the same cache manager and catalog
state.
**Why this works:** The restoration session carries the mode for query
analysis and execution while the caller SparkSession SQLConf is never changed;
descriptors retain prior storage levels.
**Scope:** Isolate per-cache scan-mode replay during V2 rename.
**Compatibility:** Direct cache variants continue to use the captured scan
semantics after rename.
**Risks:** A cloned session must retain catalog visibility and SharedState
cache ownership while isolating SQLConf.
**Constraints:** Preserve the mode captured by each descriptor rather than
rebinding it from the mutating caller.
**Success:** Concurrent operations on the caller session never observe a
rename-restoration-only CHAR/VARCHAR configuration. A concurrent caller
configuration update is not overwritten when cache restoration finishes. Each
restored cache keeps its captured mode and storage level.
##########
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:
**Non-blocking (P2):** Every read in this case enables
CONVERT_METASTORE_ORC, so the assertions cover converted LogicalRelation caches
only. Please add a conversion-disabled path that verifies native
HiveTableRelation instances keep the two bound cache identities distinct.
##########
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:
**Non-blocking (P2):** FakeV1DataSource has only a LONG column, so changing
these flags never binds either CHAR/VARCHAR scan mode. A cache regression that
refreshes only unbound relations would leave this test green; the fixture
should create both bound variants before one save.
--
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]
