srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4138245871
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -416,26 +703,30 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
*
* @return the refreshed plan if refresh succeeds, None otherwise
*/
- private def tryRefreshPlan(spark: SparkSession, plan: LogicalPlan):
Option[LogicalPlan] = {
+ private def tryRefreshPlan(plan: LogicalPlan): Option[LogicalPlan] = {
try {
- EliminateSubqueryAliases(plan) match {
- case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if
r.timeTravelSpec.isEmpty =>
- val table = CatalogV2Util.getTable(catalog, ident, options =
r.options)
- if (r.table.id == table.id) {
- Some(DataSourceV2Relation.create(table, Some(catalog),
Some(ident), r.options))
- } else {
- None
- }
- case _ =>
- Some(V2TableRefreshUtil.refresh(spark, plan))
- }
+ // Keep scan mode and any analyzer padding Project. Load Table from the
catalog rather
+ // than the shared relation cache so a recache after write sees
committed rows.
+ Some(refreshV2RelationTablesFromCatalog(plan))
} catch {
case NonFatal(e) =>
logWarning(log"Failed to refresh plan while attempting to recache", e)
None
}
}
+ // Load from the catalog, not the shared relation cache, so a recache after
write sees
+ // the committed rows instead of the copy pinned when the cache was created.
+ // Keep output types, scan mode, and padding Projects:
DataSourceV2Relation.create rewrites
+ // CHAR/VARCHAR to annotated STRING, so a later analysis pass cannot restore
the cache key.
+ private def refreshV2RelationTablesFromCatalog(plan: LogicalPlan):
LogicalPlan = {
+ plan.transformWithSubqueries {
+ case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if
r.timeTravelSpec.isEmpty =>
+ val table = CatalogV2Util.getTable(catalog, ident, options = r.options)
+ if (r.table.id == table.id) r.copy(table = table) else r
Review Comment:
`r.copy(table = table)` keeps the old output, so a direct `CACHE TABLE`
never adopts an added column and cannot rebuild after a removed column; the
existing SPARK-54424 run now fails 2/4 cases. This also bypasses
`V2TableRefreshUtil`'s table-ID, column, and metadata validation, while `else
r` silently recaches the old table after an identity change. Please restore the
distinction between direct caches—which need output reconstructed from the
fresh table—and derived plans, which retain captured output through shared V2
refresh validation.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -389,20 +602,94 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
private def tryRebuildCacheEntry(spark: SparkSession, cd: CachedData):
Option[CachedData] = {
- val sessionWithConfigsOff = getOrCloneSessionWithConfigsOff(spark)
- sessionWithConfigsOff.withActive {
- tryRefreshPlan(sessionWithConfigsOff, cd.plan).map { refreshedPlan =>
- val qe = QueryExecution.create(
- sessionWithConfigsOff,
- refreshedPlan,
- refreshPhaseEnabled = false)
- val newKey = qe.normalized
- val newCache = InMemoryRelation(cd.cachedRepresentation.cacheBuilder,
qe)
- cd.copy(plan = newKey, cachedRepresentation = newCache)
+ val policy = cacheRebuildPolicy(cd.plan)
+ val rebuildSession = sessionForCacheRebuild(spark, policy)
+ rebuildSession.withActive {
+ try {
+ tryRefreshPlan(cd.plan).map { refreshedPlan =>
+ val qe = QueryExecution.create(
+ rebuildSession,
+ refreshedPlan,
+ refreshPhaseEnabled = false)
+ // Keep the captured mode and padding Project as the cache key.
qe.normalized re-runs
+ // analysis, which can drop or rebuild that Project under the caller
session.
+ val newKey = QueryExecution.normalize(rebuildSession, refreshedPlan)
+ val newCache =
InMemoryRelation(cd.cachedRepresentation.cacheBuilder, qe)
+ cd.copy(plan = newKey, cachedRepresentation = newCache)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to rebuild cache entry while attempting to
recache", e)
+ None
+ }
+ }
+ }
+
+ // Clone only the configs-off session when a mode must be applied, so
catalog plugins stay
+ // shared with the caller. CHAR/VARCHAR confs are set on that clone, never
on `spark`.
+ private def sessionForCacheRebuild(
+ spark: SparkSession,
+ policy: CacheRebuildPolicy): SparkSession = {
+ val base = getOrCloneSessionWithConfigsOff(spark)
+ // Mixed bound modes must not share one configure() or one padding flag.
+ if (policy.mixedModes || (policy.mode.isEmpty && !policy.readSidePadding))
{
+ base
+ } else {
+ val session = if (base eq spark) spark.cloneSession() else base
+ policy.mode.foreach { m =>
+ CharVarcharScanMode.configure(
+ session.sessionState.conf,
+ m,
+ nativeCharVarcharTypes = policy.nativeTypes)
}
+ session.sessionState.conf.setConfString(
+ SQLConf.READ_SIDE_CHAR_PADDING.key,
+ policy.readSidePadding.toString)
Review Comment:
`readSidePadding` is an OR over the whole plan, while `mixedModes` only
detects different scan modes. A composite cache can contain padded and bare
PreserveNative subtrees; applying this global flag makes analysis wrap the
formerly bare sibling during rebuild, changing its semantics and cache
identity. Please carry padding per subtree, or detect mixed padding and avoid
global rebinding, with a composite regression.
--
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]
