srielau commented on code in PR #58317:
URL: https://github.com/apache/spark/pull/58317#discussion_r4138376527
##########
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:
Direct caches now split on whether the scan is CHAR/VARCHAR.
Non-CHAR scans rebuild output with `DataSourceV2Relation.create` after a
catalog `getTable`, so added/removed columns are adopted (SPARK-54424). A
table-id mismatch returns `None` instead of silently recaching the old table.
Derived plans still go through `V2TableRefreshUtil` (identity, column, and
metadata validation, captured output).
CHAR/VARCHAR scans keep captured output types: `create()` rewrites them to
annotated STRING and a later analysis pass cannot restore the cache key. That
path still refuses an identity change.
##########
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:
`CacheRebuildPolicy` now tracks mixed padding (a padded CHAR/VARCHAR subtree
plus a bare sibling) in the same walk as scan modes. Mixed padding skips the
global `READ_SIDE_CHAR_PADDING` rebind the same way mixed modes skip
`configure()`, and recache keeps the captured padding Project on the padded
side.
Added `composite PreserveNative cache keeps mixed read-side padding` in
`CachedTableSuite` (CHAR join VARCHAR, recache under padding disabled).
--
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]
